diff --git a/backend/alembic/versions/20260909_01_holding_items.py b/backend/alembic/versions/20260909_01_holding_items.py new file mode 100644 index 0000000..f356a1e --- /dev/null +++ b/backend/alembic/versions/20260909_01_holding_items.py @@ -0,0 +1,36 @@ +"""holding_items：持仓股（手动标记进「持仓」分类） + +Revision ID: 20260909_01 +Revises: 20260907_02 +Create Date: 2026-09-09 +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "20260909_01" +down_revision: Union[str, Sequence[str], None] = "20260907_02" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + "holding_items", + sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column("user_id", sa.BigInteger(), nullable=False), + sa.Column("ts_code", sa.String(length=12), nullable=False), + sa.Column("created_at", sa.DateTime(), nullable=False), + sa.ForeignKeyConstraint(["user_id"], ["users.id"], ondelete="CASCADE"), + sa.PrimaryKeyConstraint("id"), + sa.UniqueConstraint("user_id", "ts_code", name="uq_holding_user_code"), + ) + op.create_index("ix_holding_items_user_id", "holding_items", ["user_id"]) + op.create_index("ix_holding_items_ts_code", "holding_items", ["ts_code"]) + + +def downgrade() -> None: + op.drop_index("ix_holding_items_ts_code", table_name="holding_items") + op.drop_index("ix_holding_items_user_id", table_name="holding_items") + op.drop_table("holding_items") diff --git a/backend/app/api/__init__.py b/backend/app/api/__init__.py new file mode 100644 index 0000000..ecd99e7 --- /dev/null +++ b/backend/app/api/__init__.py @@ -0,0 +1,30 @@ +"""HTTP 路由（OpenAPI 契约的载体）——按域拆分： + + stocks.py /api/stocks*（列表/筛选项/公司/财务/分红/参考数据）+ /api/data/sync + etfs.py /api/etfs + /api/etf/sync* + market.py /api/market/*（总览/打板/概念板块/指数 K 线与详情） + backtest.py /api/backtest + /api/backtest/event + screener.py /api/screener/*（选股/历史/同步/个股预览） + user.py /api/preferences + /api/watchlist* + /api/trades* + _deps.py 共享件：JSON 直返缓存、复权换算、行转 Bar、共享常量与 SQL + +统一约定：prefix=/api 与 require_user 鉴权在本层挂一次，子路由不带前缀； +路由注册顺序 = include 顺序（各域路径前缀互不重叠，顺序不影响匹配）。 +""" +from fastapi import APIRouter, Depends + +from ..auth import require_user +from .backtest import router as backtest_router +from .etfs import router as etfs_router +from .market import router as market_router +from .screener import router as screener_router +from .stocks import router as stocks_router +from .user import router as user_router + +router = APIRouter(prefix="/api", dependencies=[Depends(require_user)]) +router.include_router(stocks_router) +router.include_router(etfs_router) +router.include_router(market_router) +router.include_router(backtest_router) +router.include_router(screener_router) +router.include_router(user_router) diff --git a/backend/app/api/_deps.py b/backend/app/api/_deps.py new file mode 100644 index 0000000..9ce1e2c --- /dev/null +++ b/backend/app/api/_deps.py @@ -0,0 +1,145 @@ +"""路由包共享件：JSON 直返缓存、复权换算、行转 Bar、共享常量与 SQL。 + +各域路由模块（stocks/etfs/market/backtest/screener/user）从这里取公共工具， +域内私有工具留在各自文件里。 +""" +from __future__ import annotations + +import bisect + +import pandas as pd +from fastapi import Response +from sqlalchemy import text + +from .. import cache +from ..domain import Bar + +# 复权模式白名单 +ADJUST_MODES = ("bfq", "qfq", "hfq") +# MA 全量集合（前端已改为本地计算 MA，后端始终返回此集合以保证缓存一致） +FULL_MA_SET = (5, 10, 20, 30, 60, 120, 250) +# 指数 K 线支持的周期（日线基底聚合） +INDEX_TIMEFRAMES = ("1d", "1w", "1M", "1y") + + +def raw_json(resp) -> str: + """pydantic-core（Rust）序列化：与 response_model 直返时的字节完全一致（紧凑分隔符、 + 非 ASCII 直出、浮点小数形式），且比 stdlib json.dumps 快。大响应（preview ~250KB） + 命中缓存时直接 Response 原样返回，跳过校验/再序列化。""" + return resp.model_dump_json() + + +async def cached_json_response(key: str) -> Response | None: + """两级缓存读（进程内 → Redis）：命中返回可直接吐给客户端的 Response。 + 存的均为序列化好的 JSON 字符串（Redis 侧 json.loads 后仍是 str），Redis 命中顺手晋级本地。""" + raw = cache.local_get(key) + if raw is None: + raw = await cache.cache_get(key) + if not isinstance(raw, str): + return None + cache.local_set(key, raw, ttl=120) + return Response(content=raw, media_type="application/json") + + +def series_to_jsonable(s: pd.Series) -> list[float | None]: + """NaN -> None（lightweight-charts 的 whitespace data，跳过指标预热期）。""" + out: list[float | None] = [] + for v in s.tolist(): + if v is None or (isinstance(v, float) and v != v): + out.append(None) + else: + out.append(float(v)) + return out + + +def rows_to_bars(rows) -> list[Bar]: + return [ + Bar( + ts=r.ts, open=r.open, high=r.high, low=r.low, close=r.close, volume=r.volume, + amount=getattr(r, "amount", None), turnover=getattr(r, "turnover", None), + ) + for r in rows + ] + + +# 信息卡一条 SQL 拿全：stock_basic 基本信息 + 「优先与行情同日、缺则最新日」的 daily_snapshot +# （LATERAL 单条替换原两条查询，语义不变：target 为 NULL 时全按最新日兜底）。 +# ETF 走 etf_basic 分支（代码前缀与股票不重叠，两分支至多一个命中）： +# 名称/上市日来自表内，市值（元）换算成万元与快照口径一致，无 PE/PB。 +INFO_SQL = text( + """ + SELECT ts_code, symbol, name, industry, area, market, list_date, + turnover_rate, pe_ttm, pb, total_mv, circ_mv + FROM ( + SELECT sb.ts_code, sb.symbol, sb.name, sb.industry, sb.area, sb.market, sb.list_date, + ds.turnover_rate, ds.pe_ttm, ds.pb, ds.total_mv, ds.circ_mv + FROM stock_basic sb + LEFT JOIN LATERAL ( + SELECT turnover_rate, pe_ttm, pb, total_mv, circ_mv + FROM daily_snapshot + WHERE ts_code = sb.ts_code + ORDER BY (trade_date = cast(:target AS timestamp)) DESC, trade_date DESC + LIMIT 1 + ) ds ON true + WHERE sb.ts_code = :code + UNION ALL + SELECT eb.ts_code, eb.symbol, eb.name, NULL, NULL, + CASE eb.exchange WHEN 'SH' THEN '沪市' ELSE '深市' END, eb.list_date, + eb.turnover_rate, NULL, NULL, + eb.total_mv / 10000.0, eb.circ_mv / 10000.0 + FROM etf_basic eb + WHERE eb.ts_code = :code + ) t + LIMIT 1 + """ +) + +# 复权因子是阶梯函数（除权日之间不变）：只取「变化点」行，把每符号 ~7000 行日级因子压到 +# 几十行（600118 仅 32 行），传输量再降两个数量级；lag 窗口在覆盖索引上走 Index Only Scan。 +# upto 传全局最新交易日（非分页）或 end 日期（分页）；窗口首行 prev 为 NULL 恒被保留（窗口基线因子）。 +FACTOR_STEP_SQL = text( + """ + SELECT trade_date, adj_factor FROM ( + SELECT trade_date, adj_factor, + lag(adj_factor) OVER (ORDER BY trade_date) AS prev + FROM adj_factor + WHERE ts_code = :code AND trade_date <= :upto + ) t + WHERE adj_factor IS DISTINCT FROM prev + ORDER BY trade_date + """ +) + + +def adjust_bars(bars: list[Bar], factors, from_mode: str, to_mode: str) -> list[Bar]: + """按复权因子把 K 线从 from_mode 换算到 to_mode（bfq/qfq/hfq）。 + + 相对不复权的乘数：bfq=1，qfq=f(t)/f(latest)，hfq=f(t)。 + 因子缺失的日期向前沿用最近因子（因子是阶梯函数，除权日之间不变）。 + """ + fd = sorted((f[0].date(), float(f[1])) for f in factors) + fdates = [d for d, _ in fd] + f_latest = fd[-1][1] + + def _f_at(d) -> float: + i = bisect.bisect_right(fdates, d) - 1 + return fd[i][1] if i >= 0 else fd[0][1] + + def _mult(mode: str, f: float) -> float: + if mode == "bfq": + return 1.0 + return f / f_latest if mode == "qfq" else f + + out: list[Bar] = [] + for b in bars: + f = _f_at(b.ts.date()) + m = _mult(to_mode, f) / _mult(from_mode, f) + out.append(Bar( + ts=b.ts, + open=round(b.open * m, 3), high=round(b.high * m, 3), + low=round(b.low * m, 3), close=round(b.close * m, 3), + volume=b.volume, + # 成交额/换手率是名义量，不随复权缩放 + amount=b.amount, turnover=b.turnover, + )) + return out diff --git a/backend/app/api/backtest.py b/backend/app/api/backtest.py new file mode 100644 index 0000000..6223330 --- /dev/null +++ b/backend/app/api/backtest.py @@ -0,0 +1,152 @@ +"""回测域路由：策略回测（旧 API，K线+指标+买卖点+净值）+ 自然语言事件回测。""" +from __future__ import annotations + +import json + +import pandas as pd +from fastapi import APIRouter, Depends, HTTPException +from sqlalchemy.ext.asyncio import AsyncSession + +from ..backtest.engine import BacktestConfig, run_backtest +from ..backtest.events import EventEngineError, run_event_backtest +from ..backtest.strategies import build_strategy +from ..data import fetcher, repository +from ..data.aggregation import bars_per_year, resample_bars +from ..data.symbols import is_etf_symbol +from ..db import get_session +from ..models import BacktestRun +from ..schemas import ( + BacktestRequest, + BacktestResponse, + CandleOut, + EquityPoint, + EventBacktestRequest, + EventBacktestResponse, + IndicatorOut, + MetricsOut, + SignalOut, +) +from ..screener.llm import parse_event_spec, ScreenerError +from ._deps import rows_to_bars, series_to_jsonable + +router = APIRouter() + + +@router.post("/backtest", response_model=BacktestResponse) +async def backtest( + req: BacktestRequest, + session: AsyncSession = Depends(get_session), +) -> BacktestResponse: + # 真实数据：本地无缓存则先拉取 + if not await fetcher.is_cached(session, req.symbol): + try: + await fetcher.sync_symbol(session, req.symbol, source="auto") + except Exception as e: # noqa: BLE001 + raise HTTPException(status_code=502, detail=f"数据拉取失败: {e}") + + # 日线为基底，聚合到请求周期 + rows = await repository.get_candles( + session, req.symbol, "1d", start=req.start, end=req.end, limit=100000 + ) + if not rows: + raise HTTPException(status_code=404, detail=f"无数据: symbol={req.symbol}") + + bars = resample_bars(rows_to_bars(rows), req.timeframe) + if len(bars) < 2: + raise HTTPException(status_code=400, detail=f"周期 {req.timeframe} 下数据不足，无法回测") + + try: + strategy = build_strategy(req.strategy, req.params) + except Exception as e: # noqa: BLE001 + raise HTTPException(status_code=400, detail=f"策略构建失败: {e}") + cfg = BacktestConfig( + initial_cash=req.initial_cash, + fast_mode=req.fast_mode, + bars_per_year=bars_per_year(req.timeframe), + is_fund=is_etf_symbol(req.symbol), # ETF 免印花税/过户费 + ) + result = run_backtest(bars, strategy, cfg) + + df: pd.DataFrame = result["df"] + m = result["metrics"] + + # 记录到回测运行注册表（可复现/可审计的基础） + session.add( + BacktestRun( + symbol=req.symbol, + strategy=req.strategy, + timeframe=req.timeframe, + params_json=json.dumps(req.params, ensure_ascii=False), + initial_cash=req.initial_cash, + total_return=m["total_return"], + max_drawdown=m["max_drawdown"], + sharpe=m["sharpe"], + num_trades=m["num_trades"], + ) + ) + await session.commit() + + candles = [ + CandleOut(ts=r["ts"], open=r["open"], high=r["high"], low=r["low"], + close=r["close"], volume=r["volume"], + amount=r["amount"] if "amount" in df.columns else None, + turnover=r["turnover"] if "turnover" in df.columns else None) + for _, r in df.iterrows() + ] + signals = [ + SignalOut(ts=f.ts, side=f.side.value, price=f.price, qty=f.qty) + for f in result["fills"] + ] + indicators = IndicatorOut( + strategy=req.strategy, + data={col: series_to_jsonable(df[col]) for col in result["indicator_cols"]}, + ) + equity = [EquityPoint(ts=t.to_pydatetime(), value=float(v)) + for t, v in result["equity"].items()] + + return BacktestResponse( + symbol=req.symbol, + timeframe=req.timeframe, + strategy=req.strategy, + candles=candles, + indicators=indicators, + signals=signals, + equity=equity, + metrics=MetricsOut(**m), + final_cash=result["final_cash"], + final_position=result["final_position"], + initial_cash=req.initial_cash, + ) + + +@router.post("/backtest/event", response_model=EventBacktestResponse) +async def backtest_event( + req: EventBacktestRequest, + session: AsyncSession = Depends(get_session), +) -> EventBacktestResponse: + """自然语言事件回测：入场条件命中 -> 次日买入 -> 持有 N 日，单股或全市场汇总统计。 + 直传 spec 则跳过 LLM（前端调参重跑）。""" + try: + spec = req.spec or await parse_event_spec(req.text) + result = await run_event_backtest( + session, spec, + ts_code=req.ts_code, + start=req.start.date() if req.start else None, + end=req.end.date() if req.end else None, + ) + except ScreenerError as e: + raise HTTPException(status_code=502, detail=str(e)) + except EventEngineError as e: + raise HTTPException(status_code=400, detail=str(e)) + except Exception as e: # noqa: BLE001 + raise HTTPException(status_code=500, detail=f"事件回测失败: {e}") + return EventBacktestResponse( + text=req.text, + spec=result["spec"], + universe=result["universe"], + start=result["start"], + end=result["end"], + stats=result["stats"], + trades=result["trades"], + total=result["total"], + ) diff --git a/backend/app/api/etfs.py b/backend/app/api/etfs.py new file mode 100644 index 0000000..b2545bb --- /dev/null +++ b/backend/app/api/etfs.py @@ -0,0 +1,133 @@ +"""ETF 域路由：全市场列表（东财快照 + candles 行情）+ 同步任务。""" +from __future__ import annotations + +from fastapi import APIRouter, Depends, HTTPException, Response +from sqlalchemy import text +from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy.sql.elements import TextClause + +from .. import cache +from ..auth import require_user +from ..config import settings +from ..data import etf_sync as etf_sync_mod +from ..db import get_session +from ..schemas import EtfListItemOut, EtfListResponse, EtfSyncRequest, EtfSyncStatus +from ._deps import cached_json_response, raw_json + +router = APIRouter() + +# ---------- ETF 列表（全市场浏览；行情走 candles 底座，规模/换手走东财快照） ---------- +# 与 /stocks 不同：成交额来自 candles 最新 bar（LATERAL），必须在分页前 join 才能参与 +# 排序 —— ETF 全市场仅 ~1100 行，3 个索引探测/行 也就几 ms，可以承受。 +# 排序列白名单（键→表达式）；order_by 由白名单拼接进模板，不接收用户原文。 +_ETFS_SORTS = { + "symbol": "eb.symbol", + "close": "c.close", + "pct_chg": "pct_chg", + "amount": "c.amount", + "total_mv": "eb.total_mv", + "circ_mv": "eb.circ_mv", + "turnover_rate": "eb.turnover_rate", +} + +_ETFS_SQL_TMPL = """ + SELECT eb.ts_code, eb.symbol, eb.name, eb.exchange, eb.list_date, + (w.id IS NOT NULL) AS watched, + eb.turnover_rate, + round((eb.total_mv / 100000000.0)::numeric, 2) AS total_mv, + round((eb.circ_mv / 100000000.0)::numeric, 2) AS circ_mv, + c.close AS close, prev.close AS prev_close, c.ts AS last_ts, + CASE WHEN c.close IS NOT NULL AND prev.close IS NOT NULL AND prev.close <> 0 + THEN round(((c.close / prev.close - 1) * 100)::numeric, 2) END AS pct_chg, + round((c.amount / 100000000.0)::numeric, 2) AS amount + FROM etf_basic eb + LEFT JOIN watchlist_items w ON w.ts_code = eb.ts_code AND w.user_id = :uid + LEFT JOIN LATERAL ( + SELECT close, ts, amount FROM candles + WHERE symbol = eb.symbol AND timeframe = '1d' + ORDER BY ts DESC LIMIT 1 + ) c ON true + LEFT JOIN LATERAL ( + SELECT close FROM candles + WHERE symbol = eb.symbol AND timeframe = '1d' AND ts < c.ts + ORDER BY ts DESC LIMIT 1 + ) prev ON c.ts IS NOT NULL + WHERE (:search = '' OR eb.symbol LIKE :psearch OR eb.name LIKE :psearch) + AND (:exchange = '' OR eb.exchange = :exchange) + AND (:watched_only = false OR w.id IS NOT NULL) + ORDER BY {order_by} + LIMIT :limit OFFSET :offset +""" + +_ETFS_COUNT_SQL = text(""" + SELECT count(*) FROM etf_basic eb + LEFT JOIN watchlist_items w ON w.ts_code = eb.ts_code AND w.user_id = :uid + WHERE (:search = '' OR eb.symbol LIKE :psearch OR eb.name LIKE :psearch) + AND (:exchange = '' OR eb.exchange = :exchange) + AND (:watched_only = false OR w.id IS NOT NULL) +""") + + +def _etfs_sql(sort: str, order: str) -> TextClause: + col = _ETFS_SORTS.get(sort, _ETFS_SORTS["symbol"]) + direction = "DESC" if order == "desc" else "ASC" + nulls = " NULLS LAST" if col != "eb.symbol" else "" # 无行情/无快照的排最后 + return text(_ETFS_SQL_TMPL.format(order_by=f"{col} {direction}{nulls}")) + + +@router.get("/etfs", response_model=EtfListResponse) +async def list_etfs( + search: str = "", + exchange: str = "", + watched_only: bool = False, + sort: str = "symbol", + order: str = "asc", + limit: int = 100, + offset: int = 0, + session: AsyncSession = Depends(get_session), + user=Depends(require_user), +) -> Response: + """全市场场内 ETF 列表：etf_basic 名称/规模（东财快照）+ candles 最新收盘/涨跌幅/成交额。 + exchange ∈ {SH, SZ}（空 = 全部）；sort ∈ {symbol,close,pct_chg,amount,total_mv,circ_mv, + turnover_rate}（白名单，其他值回落 symbol），order ∈ asc/desc；快照/行情列排序时 + 缺失值恒排末尾。缓存：按「用户自选版本 + etf 版本 + 查询参数」缓存整页， + ETF 同步完成（bump ver:etf / ver:candles）或自选增删即失效。 + """ + search = search.strip() + sort = sort if sort in _ETFS_SORTS else "symbol" + order = "desc" if order.lower() == "desc" else "asc" + limit = max(1, min(limit, 500)) + offset = max(0, offset) + key = ( + f"etfsj:u{user.id}" + f":v{await cache.get_version(f'watchlist:{user.id}')}" + f":v{await cache.get_version('etf')}" + f":{cache.digest(search, exchange, watched_only, sort, order, limit, offset)}" + ) + cached = await cached_json_response(key) + if cached is not None: + return cached + params = { + "search": search, "psearch": f"%{search}%", + "exchange": exchange.upper(), "watched_only": watched_only, + "uid": user.id, "limit": limit, "offset": offset, + } + total = (await session.execute(_ETFS_COUNT_SQL, params)).scalar_one() + rows = (await session.execute(_etfs_sql(sort, order), params)).mappings().all() + resp = EtfListResponse(total=total, items=[EtfListItemOut(**r) for r in rows]) + raw = raw_json(resp) + cache.local_set(key, raw, ttl=min(120, settings.stocks_cache_ttl)) + cache.set_bg(key, raw, ttl=settings.stocks_cache_ttl) + return Response(content=raw, media_type="application/json") + + +@router.post("/etf/sync", response_model=EtfSyncStatus) +async def etf_sync_start(req: EtfSyncRequest) -> EtfSyncStatus: + """启动全市场 ETF 同步（后台任务：东财快照 -> etf_basic，逐只日线 -> candles）。""" + return EtfSyncStatus(**await etf_sync_mod.start_sync(full=req.full)) + + +@router.get("/etf/sync/status", response_model=EtfSyncStatus) +async def etf_sync_status(session: AsyncSession = Depends(get_session)) -> EtfSyncStatus: + """ETF 同步任务状态与数据实况（ETF 数 / 最新交易日）。""" + return EtfSyncStatus(**await etf_sync_mod.get_status(session)) diff --git a/backend/app/api/market.py b/backend/app/api/market.py new file mode 100644 index 0000000..a60ce2b --- /dev/null +++ b/backend/app/api/market.py @@ -0,0 +1,252 @@ +"""行情专题路由：大盘总览 / 打板 / 概念板块 / 指数（上证 K 线、国际指数、指数详情与权重）。""" +from __future__ import annotations + +from datetime import datetime + +from fastapi import APIRouter, Depends, HTTPException, Response +from pydantic import TypeAdapter +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from .. import cache +from ..data import index_global as index_mod +from ..data import limit_board as limit_board_mod +from ..data import ths_board as ths_board_mod +from ..data.index_series import SH_INDEX, get_index_daily +from ..data.aggregation import resample_bars +from ..data.limit_board import LimitBoardError +from ..data.market_overview import MarketOverviewError, fetch_overview +from ..data.ths_board import ThsBoardError +from ..db import async_session, get_session +from ..models import StockBasic, TradeCalendar +from ..schemas import ( + CandleOut, + GlobalIndexListResponse, + IndexBasicOut, + IndexDetailResponse, + IndexQuoteBriefOut, + IndexValuationPointOut, + IndexWeightItemOut, + IndexWeightsResponse, + LimitBoardResponse, + MarketOverviewResponse, + ThsBoardListResponse, + ThsBoardMembersResponse, +) +from ._deps import INDEX_TIMEFRAMES, cached_json_response + +router = APIRouter() + + +async def _is_trading_day_today() -> bool | None: + """今日是否 A 股交易日（trade_date 为 String(8) unique 索引，等值查亚毫秒级）； + DB 不可用时返回 None，调用方回退 weekday 启发式（只影响盘中 TTL 精度）。""" + today8 = datetime.now().strftime("%Y%m%d") + try: + async with async_session() as s: + return bool(await s.scalar( + select(TradeCalendar.id).where(TradeCalendar.trade_date == today8).limit(1) + )) + except Exception: # noqa: BLE001 + return None + + +@router.get("/market/overview", response_model=MarketOverviewResponse) +async def get_market_overview(session: AsyncSession = Depends(get_session)) -> MarketOverviewResponse: + """主页大盘总览：A 股 + 港美指数实时价（腾讯）叠加收盘历史走势（tushare）， + 沪深两市市值/成交统计 + 成交额历史。部分来源失败不影响其余。 + """ + is_trading_day: bool | None = None + try: + is_trading_day = bool(await session.scalar( + select(TradeCalendar.id).where( + TradeCalendar.trade_date == datetime.now().strftime("%Y%m%d")).limit(1) + )) + except Exception: # noqa: BLE001 —— 判定失败只影响「今日盘中 bar」是否追加 + pass + try: + data = await fetch_overview(is_trading_day=is_trading_day) + except MarketOverviewError as e: + raise HTTPException(status_code=503, detail=str(e)) from e + return MarketOverviewResponse(**data) + + +@router.get("/market/limit-board", response_model=LimitBoardResponse) +async def get_limit_board(session: AsyncSession = Depends(get_session)) -> LimitBoardResponse: + """首页打板专题（同花顺口径）：涨停/炸板/跌停三池 + 连板天梯 + 涨停最强板块， + 当日快照（盘中 5 分钟 / 盘后 4 小时，整包 SWR 缓存）。部分池失败不影响其余。""" + is_trading_day: bool | None = None + try: + is_trading_day = bool(await session.scalar( + select(TradeCalendar.id).where( + TradeCalendar.trade_date == datetime.now().strftime("%Y%m%d")).limit(1) + )) + except Exception: # noqa: BLE001 —— 判定失败回退 weekday 启发式（影响盘中 TTL 精度而已） + pass + try: + data = await limit_board_mod.fetch_limit_board(is_trading_day) + except LimitBoardError as e: + raise HTTPException(status_code=503, detail=str(e)) from e + return LimitBoardResponse(**data) + + +@router.get("/market/boards", response_model=ThsBoardListResponse) +async def list_ths_boards() -> ThsBoardListResponse: + """概念/行业板块列表（同花顺口径，全部类型一次给全，前端本地过滤）： + ths_index 列表直缓存 24h + ths_daily 当日快照 SWR（盘中 5 分钟 / 盘后 4 小时）。""" + try: + data = await ths_board_mod.fetch_boards(await _is_trading_day_today()) + except ThsBoardError as e: + raise HTTPException(status_code=503, detail=str(e)) from e + return ThsBoardListResponse(**data) + + +@router.get("/market/boards/{code}/members", response_model=ThsBoardMembersResponse) +async def list_ths_board_members(code: str, session: AsyncSession = Depends(get_session)) -> ThsBoardMembersResponse: + """板块成分股（ths_member 懒加载缓存 24h）+ 最新现价/涨跌幅（candles LATERAL 现算）。""" + bc = code.strip().upper() + try: + boards = await ths_board_mod.get_board_list() + except ThsBoardError as e: + raise HTTPException(status_code=503, detail=str(e)) from e + board = next((b for b in boards if b["ts_code"] == bc), None) + if board is None: + raise HTTPException(status_code=404, detail=f"未知板块: {bc}") + try: + members = await ths_board_mod.get_members(session, bc) + except Exception: + raise HTTPException(status_code=503, detail="板块成分拉取失败，请稍后重试") + return ThsBoardMembersResponse(code=bc, name=board.get("name"), members=members) + + +@router.get("/market/index-candles", response_model=list[CandleOut]) +async def get_index_candles(timeframe: str = "1d") -> Response: + """上证指数全量 K 线：日线为基底（tushare index_daily，进程内+Redis 缓存）， + 聚合到 1d/1w/1M/1y。收盘口径（数据随 EOD 更新，与总览 spark 一致）。""" + if timeframe not in INDEX_TIMEFRAMES: + raise HTTPException(status_code=400, detail=f"timeframe 仅支持 {'/'.join(INDEX_TIMEFRAMES)}") + key = f"idxkj:{cache.digest('idxc', SH_INDEX, timeframe)}" + cached = await cached_json_response(key) + if cached is not None: + return cached + try: + bars = resample_bars(await get_index_daily(), timeframe) + except Exception as e: # noqa: BLE001 + raise HTTPException(status_code=502, detail=f"指数数据获取失败: {e}") + outs = [ + CandleOut(ts=b.ts, open=b.open, high=b.high, low=b.low, close=b.close, + volume=b.volume, amount=b.amount, turnover=None) + for b in bars + ] + # pydantic-core 序列化（与 raw_json 同款），历史不可变、TTL 兜到当日更新 + raw = TypeAdapter(list[CandleOut]).dump_json(outs).decode() + cache.local_set(key, raw, ttl=300) + await cache.cache_set(key, raw, ttl=7200) + return Response(content=raw, media_type="application/json") + + +@router.get("/market/global-indexes", response_model=GlobalIndexListResponse) +async def get_global_indexes() -> GlobalIndexListResponse: + """国际指数卡片列表（index_global 21 个指数最新收盘 + 45 日 spark，SWR 缓存）。""" + try: + data = await index_mod.fetch_global_list() + except index_mod.GlobalIndexError as e: + raise HTTPException(status_code=503, detail=str(e)) from e + payload = dict(data) + payload["updated_at"] = payload.pop("fetched_at") + return GlobalIndexListResponse(**payload) + + +def _index_or_404(code: str) -> str: + """详情/K线/权重接口只放行白名单内的指数 code。""" + if not index_mod.ensure_known(code): + raise HTTPException(status_code=404, detail=f"不支持的指数代码: {code}") + return code + + +@router.get("/market/indexes/{code}", response_model=IndexDetailResponse) +async def get_index_detail(code: str) -> IndexDetailResponse: + """指数详情聚合：最新行情（收盘口径）+ 基本信息（国内 index_basic / 国际静态表） + + 估值指标（index_dailybasic，仅部分国内指数）。各层自带缓存，直接组装。""" + code = _index_or_404(code) + if index_mod.is_cn_index(code): + name, region = index_mod.CN_INDEXES.get(code, code), "cn" + else: + g = index_mod.GLOBAL_META[code] + name, region = g["name"], g["region"] + + try: + quote = await index_mod.fetch_index_quote(code) + except index_mod.GlobalIndexError as e: + raise HTTPException(status_code=502, detail=f"指数行情获取失败: {e}") from e + + basic_raw = await index_mod.get_index_basic(code) + basic = IndexBasicOut(**basic_raw) if basic_raw else None + + valuation_rows = await index_mod.get_index_valuation(code) + valuation = IndexValuationPointOut(**valuation_rows[-1]) if valuation_rows else None + history = [IndexValuationPointOut(**r) for r in valuation_rows] + + return IndexDetailResponse( + code=code, name=name, region=region, + quote=IndexQuoteBriefOut(**quote), + basic=basic, valuation=valuation, valuation_history=history, + ) + + +@router.get("/market/indexes/{code}/candles", response_model=list[CandleOut]) +async def get_any_index_candles(code: str, timeframe: str = "1d") -> Response: + """白名单指数全量 K 线（国内 index_daily / 国际 index_global），1d/1w/1M/1y 聚合。 + 收盘口径，历史不可变、TTL 兜到当日更新（与首页上证 K 线同款缓存策略）。""" + code = _index_or_404(code) + if timeframe not in INDEX_TIMEFRAMES: + raise HTTPException(status_code=400, detail=f"timeframe 仅支持 {'/'.join(INDEX_TIMEFRAMES)}") + key = f"idxck:{cache.digest('idxk2', code, timeframe)}" + cached = await cached_json_response(key) + if cached is not None: + return cached + try: + bars = resample_bars(await index_mod.get_index_bars(code), timeframe) + except Exception as e: # noqa: BLE001 + raise HTTPException(status_code=502, detail=f"指数数据获取失败: {e}") + outs = [ + CandleOut(ts=b.ts, open=b.open, high=b.high, low=b.low, close=b.close, + volume=b.volume, amount=b.amount, turnover=None) + for b in bars + ] + raw = TypeAdapter(list[CandleOut]).dump_json(outs).decode() + cache.local_set(key, raw, ttl=300) + await cache.cache_set(key, raw, ttl=7200) + return Response(content=raw, media_type="application/json") + + +@router.get("/market/indexes/{code}/weights", response_model=IndexWeightsResponse) +async def get_index_weights( + code: str, + limit: int = 50, + session: AsyncSession = Depends(get_session), +) -> IndexWeightsResponse: + """指数成分股权重（index_weight 最近月度快照，按权重降序取前 limit）。 + 仅国内指数有数据；成分股名称从本地 stock_basic 回填。""" + code = _index_or_404(code) + limit = max(1, min(limit, 300)) + data = await index_mod.get_index_weights(code) + if data is None: + raise HTTPException(status_code=404, detail=f"该指数暂无成分权重数据: {code}") + + items = data["items"][:limit] + codes = [it["con_code"] for it in items] + names: dict[str, str] = {} + if codes: + try: + rows = await session.execute( + select(StockBasic.ts_code, StockBasic.name).where(StockBasic.ts_code.in_(codes)) + ) + names = {r[0]: r[1] for r in rows.all()} + except Exception: # noqa: BLE001 —— 名称缺失不阻塞权重展示 + pass + return IndexWeightsResponse( + trade_date=data["trade_date"], total=data["total"], + items=[IndexWeightItemOut(con_code=c, name=names.get(c), weight=w) + for c, w in ((it["con_code"], it["weight"]) for it in items)], + ) diff --git a/backend/app/api/screener.py b/backend/app/api/screener.py new file mode 100644 index 0000000..fc06caf --- /dev/null +++ b/backend/app/api/screener.py @@ -0,0 +1,422 @@ +"""智能选股域路由：自然语言选股（NDJSON 流式）+ 提问历史 + 全市场同步 + 个股预览。""" +from __future__ import annotations + +import asyncio +import json +from datetime import datetime + +import pandas as pd +from fastapi import APIRouter, Depends, HTTPException, Response +from fastapi.responses import StreamingResponse +from sqlalchemy import func, select, text +from sqlalchemy.ext.asyncio import AsyncSession + +from .. import cache +from .. import indicators as ind +from ..auth import require_user +from ..config import settings +from ..data import fetcher, repository +from ..data.aggregation import resample_bars +from ..data.symbols import plain_code +from ..db import async_session, get_session +from ..models import AdjFactor, Candle, ScreenerQuery +from ..schemas import ( + CandleOut, + PreviewInfoOut, + PreviewResponse, + ScreenerQueryListResponse, + ScreenerQueryOut, + ScreenerRunRequest, + ScreenerRunResponse, + ScreenerSyncRequest, + ScreenerSyncStatus, +) +from ..screener import engine, market_sync +from ..screener.engine import DataNotReadyError +from ..screener.llm import ScreenerError, parse_conditions +from ._deps import ( + ADJUST_MODES, + FACTOR_STEP_SQL, + FULL_MA_SET, + INFO_SQL, + adjust_bars, + cached_json_response, + raw_json, + rows_to_bars, + series_to_jsonable, +) + +router = APIRouter() + + +# ---------- 智能选股 ---------- +@router.post("/screener/run") +async def screener_run( + req: ScreenerRunRequest, + session: AsyncSession = Depends(get_session), + user=Depends(require_user), +) -> StreamingResponse: + """自然语言 -> LLM 解析条件 -> 全市场筛选。也可直传 conditions 跳过 LLM（微调再跑）。 + + NDJSON 流式响应（每行一个 JSON 事件，前端逐行渲染进度）： + {"type":"stage","key":"llm|date|prefilter|bars|filter_done|done","msg":"…","ms":123} + {"type":"parsed","conditions":{…},"ms":456} LLM 解析出的结构化条件 + {"type":"candidates","count":5400,"msg":"…","ms":…} SQL 预筛后的候选数 + {"type":"progress","done":500,"total":5400} 逐股指标过滤进度 + {"type":"result","result":{…ScreenerRunResponse…},"ms":…} + {"type":"error","message":"…","code":400} 流中途失败（HTTP 已 200） + 成功的提问（含解析出的条件与命中数）记录到 screener_queries，供历史一键重跑。 + """ + limit = settings.screener_default_limit + + async def gen(): + try: + if req.conditions: + conds = req.conditions + else: + yield _ndjson({"type": "stage", "key": "llm", + "msg": f"AI 解析条件中（{settings.llm_model}）…"}) + conds = await parse_conditions(req.text) + if not conds.indicator and not conds.snapshot: + yield _ndjson({"type": "error", "code": 400, + "message": "AI 未从描述中解析出任何筛选条件，请换种说法"}) + return + yield _ndjson({"type": "parsed", "conditions": conds.model_dump()}) + + result = None + async for ev in engine.run_screen_events(session, conds, limit): + if ev["type"] == "result": + result = ev["result"] + yield _ndjson({"type": "stage", "key": "done", "ms": ev.get("ms"), + "msg": f"筛选完成：{result['total']} 只命中（数据基准 {result['trade_date']:%Y-%m-%d}）"}) + else: + yield _ndjson(ev) + + if result is None: + yield _ndjson({"type": "error", "code": 500, "message": "选股流程未产出结果"}) + return + yield _ndjson({"type": "result", "result": ScreenerRunResponse(**result).model_dump(mode="json")}) + + # 相同文本 + 相同条件的上一条不重复记录（一键重跑场景） + exists = ( + await session.execute( + select(ScreenerQuery.id).where( + ScreenerQuery.user_id == user.id, + ScreenerQuery.text == req.text.strip(), + ScreenerQuery.conditions_json == json.dumps(conds.model_dump(), ensure_ascii=False), + ) + ) + ).scalar_one_or_none() + if exists is None: + session.add(ScreenerQuery( + user_id=user.id, + text=req.text.strip(), + conditions_json=json.dumps(conds.model_dump(), ensure_ascii=False), + hit_count=result.get("total", 0), + )) + await session.commit() + except DataNotReadyError as e: + yield _ndjson({"type": "error", "code": 409, "message": str(e)}) + except ValueError as e: # 未知指标/字段、条件为空 + yield _ndjson({"type": "error", "code": 400, "message": str(e)}) + except ScreenerError as e: + code = 503 if "未配置 LLM_API_KEY" in str(e) else 502 + yield _ndjson({"type": "error", "code": code, "message": str(e)}) + except Exception as e: # noqa: BLE001 + yield _ndjson({"type": "error", "code": 500, "message": f"选股失败: {e}"}) + + return StreamingResponse(gen(), media_type="application/x-ndjson", + headers={"Cache-Control": "no-store", "X-Accel-Buffering": "no"}) + + +def _ndjson(obj: dict) -> str: + """dict -> NDJSON 行（json.dumps 保证 default=str 兜底 datetime 等）。""" + return json.dumps(obj, ensure_ascii=False, default=str) + "\n" + + +@router.get("/screener/queries", response_model=ScreenerQueryListResponse) +async def screener_queries( + limit: int = 20, + session: AsyncSession = Depends(get_session), + user=Depends(require_user), +) -> ScreenerQueryListResponse: + """当前用户的提问历史（最新在前，含解析出的条件与命中数，可一键重跑）。""" + limit = max(1, min(limit, 100)) + rows = ( + await session.execute( + select(ScreenerQuery) + .where(ScreenerQuery.user_id == user.id) + .order_by(ScreenerQuery.created_at.desc()) + .limit(limit) + ) + ).scalars().all() + items = [] + for r in rows: + conds = None + if r.conditions_json: + try: + from ..schemas import ScreenConditions + conds = ScreenConditions.model_validate_json(r.conditions_json) + except Exception: # noqa: BLE001 —— 旧格式/解析失败则只展示文本 + conds = None + items.append(ScreenerQueryOut( + id=r.id, text=r.text, conditions=conds, hit_count=r.hit_count, created_at=r.created_at + )) + return ScreenerQueryListResponse(items=items) + + +@router.delete("/screener/queries/{query_id}", status_code=204) +async def screener_query_delete( + query_id: int, + session: AsyncSession = Depends(get_session), + user=Depends(require_user), +) -> None: + await session.execute( + text("DELETE FROM screener_queries WHERE id = :i AND user_id = :u"), + {"i": query_id, "u": user.id}, + ) + await session.commit() + + +# ---------- 全市场数据同步 ---------- +@router.post("/screener/sync", response_model=ScreenerSyncStatus) +async def screener_sync_start( + req: ScreenerSyncRequest, session: AsyncSession = Depends(get_session) +) -> ScreenerSyncStatus: + """启动全市场数据同步（后台任务，立即返回状态）。""" + try: + await market_sync.start_sync(session, req.days, req.force) + except ScreenerError as e: + raise HTTPException(status_code=503, detail=str(e)) + status = await market_sync.get_sync_status(session) + return ScreenerSyncStatus(**{k: status.get(k) for k in ScreenerSyncStatus.model_fields}) + + +@router.get("/screener/sync/status", response_model=ScreenerSyncStatus) +async def screener_sync_status(session: AsyncSession = Depends(get_session)) -> ScreenerSyncStatus: + """同步任务状态 + 数据实况（最新交易日/行数/ready）。""" + status = await market_sync.get_sync_status(session) + return ScreenerSyncStatus(**{k: status.get(k) for k in ScreenerSyncStatus.model_fields}) + + +# ---------- 个股详情预览 ---------- +@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", + zx: str = "10,20,30,60", end: str | None = None, + session: AsyncSession = Depends(get_session), +) -> Response: + """个股详情预览：日线（candles 不复权底座 + adj_factor 本地换算 bfq/qfq/hfq， + 未缓存自动拉取，落后全市场最新交易日则强制刷新）+ 全套指标 + 最新截面信息卡。 + timeframe 聚合到周/月/年（先复权再聚合）；mas 指定主图 MA 周期（逗号分隔）。 + end=YYYY-MM-DD 时为「向前翻页」：返回该日之前最近 limit 根（含预热计算指标）， + has_more 标记窗口前是否还有更早历史，前端据此继续向左滚动加载。""" + if adjust not in ADJUST_MODES: + raise HTTPException(status_code=400, detail=f"adjust 仅支持 {'/'.join(ADJUST_MODES)}") + if timeframe not in ("1d", "1w", "1M", "1y"): + raise HTTPException(status_code=400, detail="timeframe 仅支持 1d/1w/1M/1y") + try: + ma_periods = sorted({int(p) for p in mas.split(",") if p.strip().isdigit() and 1 <= int(p) <= 500}) + except ValueError: + 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: + try: + end_dt = datetime.strptime(end.strip()[:10], "%Y-%m-%d") + except ValueError: + raise HTTPException(status_code=400, detail="end 格式应为 YYYY-MM-DD") + symbol = plain_code(ts_code) + + # --- 两级读缓存：历史窗口（end 翻页）只增不改，最新窗口每日由全市场同步推进； + # 键含 ver:candles 版本号（同步完成后自增，旧缓存全部失效），TTL 兜底（cache.py）。 + # 存序列化好的 JSON 直返（j: 前缀），跳过 json.loads + pydantic 校验/序列化（热路径数百 ms → 个位数）。 + # 注：ma_periods 不参与缓存键 —— 前端已改为本地计算 MA，后端始终返回全量 MA 集合 + cache_key = cache.digest( + "preview", ts_code, timeframe, limit, adjust, + end_dt.strftime("%Y-%m-%d") if end_dt else None, + await cache.get_version("candles"), + ) + cached = await cached_json_response(f"pvj:{cache_key}") + if cached is not None: + return cached + + # --- 日线：candles(全量不复权底座)；未缓存拉取，落后于全市场最新交易日则强制刷新 --- + # fetcher 只做「不复权」增量 upsert，底座口径恒为 bfq（TDX 全量 + Tushare 增量）， + # 复权（qfq/hfq）读取时按 adj_factor 表本地换算。 + # 每次只取「窗口 + 400 根预热」行（MA250/MACD EMA 在 400 根内充分收敛），不拉全量： + # 首屏 ~500 根秒开，向左滚动时按 end 参数逐页向前翻。 + frame_mult = {"1d": 1, "1w": 6, "1M": 24, "1y": 280}[timeframe] + fetch_n = min(100000, limit * frame_mult + 400) + source = "bfq" + mode = "bfq" + # 并发约定：注入 session 与 s2 各占一条连接，每次 gather 里每个 session 恰好跑一条查询 + # （AsyncSession 单连接非并发安全），把 ~6 次串行 DB RTT 折叠成 2 个波次。 + async with async_session() as s2: + if end_dt is not None: + # 向前翻页：取 end 之前的历史窗口，不触发同步（历史浏览）；max(ts) 用不到 + rows = await repository.get_candles_before(session, symbol, "1d", before=end_dt, limit=fetch_n) + global_latest = None + else: + # Wave 1：candles 窗口（注入 session）+ 全市场最新交易日（s2）并行 + rows, global_latest = await asyncio.gather( + repository.get_recent_candles(session, symbol, "1d", limit=fetch_n), + s2.scalar(select(func.max(Candle.ts)).where(Candle.timeframe == "1d")), + ) + try: + if not rows: + await fetcher.sync_symbol(session, symbol, source="auto") + rows = await repository.get_recent_candles(session, symbol, "1d", limit=fetch_n) + 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） + await session.rollback() + if not rows: + rows = [] + bars = rows_to_bars(rows) + + 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，前端停止向前翻页 + + # --- Wave 2：复权因子（s2，覆盖索引 Index Only Scan）+ 信息卡（注入 session，LATERAL 一条）并行 --- + async def _fetch_factors() -> list | None: + if adjust == mode or not bars: + return None + # 只取因子「变化点」行（覆盖索引 Index Only Scan，免堆访问——adj_factor 堆碎片化 + # 严重）；bisect 在阶梯函数上取值与日级序列逐字节一致 + if end_dt is not None: + # 分页：窗口 ≤ end 的变化点 + 全局最新因子（qfq 以最新因子归一） + win = list((await s2.execute(FACTOR_STEP_SQL, {"code": ts_code, "upto": end_dt})).all()) + if win: + latest_f = (await s2.execute( + select(AdjFactor.trade_date, AdjFactor.adj_factor) + .where(AdjFactor.ts_code == ts_code) + .order_by(AdjFactor.trade_date.desc()).limit(1) + )).first() + if latest_f is not None: + win.append(latest_f) + return win or None + # 非分页：上界 global_latest（≥ 最新 bar），末项变化点即全局最新因子，比 + # 「窗口 ≤ bars[-1].ts + 单独 latest」少一次查询 + return list((await s2.execute( + FACTOR_STEP_SQL, {"code": ts_code, "upto": global_latest} + )).all()) or None + + factors, info_row = await asyncio.gather( + _fetch_factors(), + session.execute(INFO_SQL, {"code": ts_code, "target": last_daily.ts if last_daily else None}), + ) + + # --- 复权换算：请求模式与底座模式不同时按 adj_factor 本地换算（无因子则维持原样） --- + if factors: + bars = adjust_bars(bars, factors, mode, adjust) + mode = adjust + source = adjust + + # --- 周期聚合：复权之后按日历聚合到周/月/年，指标在聚合后的序列上计算 --- + bars = resample_bars(bars, timeframe) + + # --- 指标（在预热窗口上计算后截尾，保证预热正确；翻页到底的空页跳过） --- + has_more = len(bars) > limit # 返回窗口之前还有更早历史（含预热行） + indicators: dict[str, dict[str, list[float | None]]] = {} + if bars: + df = pd.DataFrame({"close": [b.close for b in bars], "high": [b.high for b in bars], "low": [b.low for b in bars]}) + closes, highs, lows = df["close"], df["high"], df["low"] + macd = ind.macd(closes) + kdj = ind.kdj(highs, lows, closes) + boll = ind.bollinger(closes) + indicators = { + # MA 始终返回全量集合（前端本地计算 MA，此处仅保留兼容；缓存键不依赖 ma_periods） + "ma": {f"ma{p}": series_to_jsonable(ind.ma(closes, p)) for p in FULL_MA_SET}, + "macd": { + "dif": series_to_jsonable(macd["macd"]), + "dea": series_to_jsonable(macd["signal"]), + "hist": series_to_jsonable(macd["hist"]), + }, + "kdj": {k: series_to_jsonable(kdj[k]) for k in ("k", "d", "j")}, + "rsi": { + "rsi6": series_to_jsonable(ind.rsi(closes, 6)), + "rsi12": series_to_jsonable(ind.rsi(closes, 12)), + "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:] + + # --- 信息卡：Wave 2 已并行取回（stock_basic + 与行情同日对齐的快照、缺则最新日，见 INFO_SQL） --- + row = info_row.first() + if row is not None: + m = row._mapping + sb_name, sb_industry, sb_area, sb_market, sb_list_date = ( + m["name"], m["industry"], m["area"], m["market"], m["list_date"] + ) + ds_turnover, ds_pe, ds_pb, ds_tmv, ds_cmv = ( + m["turnover_rate"], m["pe_ttm"], m["pb"], m["total_mv"], m["circ_mv"] + ) + else: + sb_name = sb_industry = sb_area = sb_market = sb_list_date = None + ds_turnover = ds_pe = ds_pb = ds_tmv = ds_cmv = None + + def _yi(v) -> float | None: + if v is None: + return None + v = float(v) + return None if v != v else round(v / 1e4, 2) # 万元 -> 亿元 + + info = PreviewInfoOut( + ts_code=ts_code, + symbol=symbol, + name=sb_name or ts_code, + industry=sb_industry, + area=sb_area, + market=sb_market, + list_date=sb_list_date, + 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, + pe_ttm=ds_pe, + pb=ds_pb, + total_mv=_yi(ds_tmv), + circ_mv=_yi(ds_cmv), + ) + + candles = [ + CandleOut(ts=b.ts, open=b.open, high=b.high, low=b.low, close=b.close, + volume=b.volume, amount=b.amount, turnover=b.turnover) + for b in bars[-limit:] + ] + resp = PreviewResponse(ts_code=ts_code, symbol=symbol, source=source, info=info, + candles=candles, indicators=indicators, has_more=has_more) + # 只序列化一次：本地（同步，120s）+ Redis（后台写，600s TTL 兜底跨进程/重启） + raw = raw_json(resp) + cache.local_set(f"pvj:{cache_key}", raw, ttl=120) + cache.set_bg(f"pvj:{cache_key}", raw, ttl=600) + return Response(content=raw, media_type="application/json") diff --git a/backend/app/api/stocks.py b/backend/app/api/stocks.py new file mode 100644 index 0000000..1f82b45 --- /dev/null +++ b/backend/app/api/stocks.py @@ -0,0 +1,283 @@ +"""股票域路由：全市场列表 + 筛选项 + 个股公司/财务/分红/参考数据懒加载 + 手动数据同步。""" +from __future__ import annotations + +from fastapi import APIRouter, Depends, HTTPException, Response +from sqlalchemy import text +from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy.sql.elements import TextClause + +from .. import cache +from ..auth import require_user +from ..config import settings +from ..data import company as company_mod +from ..data import dividend as dividend_mod +from ..data import finance as finance_mod +from ..data import reference as reference_mod +from ..data import fetcher +from ..data.symbols import is_etf_symbol, to_ts_code +from ..db import get_session +from ..schemas import ( + FacetItemOut, + StockCompanyOut, + StockDividendOut, + StockFacetsResponse, + StockFinanceOut, + StockListItemOut, + StockListResponse, + StockReferenceOut, + SyncRequest, + SyncResponse, +) +from ._deps import cached_json_response, raw_json + +router = APIRouter() + + +@router.post("/data/sync", response_model=SyncResponse) +async def sync_data(req: SyncRequest, session: AsyncSession = Depends(get_session)) -> SyncResponse: + """主动拉取并缓存某标的的日线（Tushare 主 -> AKShare 兜底）。""" + try: + res = await fetcher.sync_symbol( + session, req.symbol, start=req.start, end=req.end, source=req.source, force=req.force + ) + return SyncResponse(**res) + except Exception as e: # noqa: BLE001 + raise HTTPException(status_code=502, detail=str(e)) + + +# ---------- 股票列表（全市场浏览） ---------- +# 过滤/排序/分页在 stock_basic+watchlist+daily_snapshot 上完成（快照按最新交易日 +# 走唯一索引 join，便宜），再对「本页」≤limit 只股票补最新价/昨收（LATERAL 扫 +# candles，贵）——旧写法对全市场 ~5000 只逐个算，每页都白算 50 倍的行情量。 +# 排序列白名单（键→CTE 内表达式）；order_by 由白名单拼接进模板，不接收用户原文。 +_STOCKS_SORTS = { + "symbol": "sb.symbol", + "total_mv": "snap.total_mv", + "circ_mv": "snap.circ_mv", + "pe_ttm": "snap.pe_ttm", + "pb": "snap.pb", + "turnover_rate": "snap.turnover_rate", +} + +_STOCKS_SQL_TMPL = """ + WITH page AS ( + SELECT sb.ts_code, sb.symbol, sb.name, sb.industry, sb.market, + (w.id IS NOT NULL) AS watched, + (h.id IS NOT NULL) AS held, + snap.turnover_rate, snap.pe_ttm, snap.pb, snap.total_mv, snap.circ_mv + FROM stock_basic sb + LEFT JOIN watchlist_items w ON w.ts_code = sb.ts_code AND w.user_id = :uid + LEFT JOIN holding_items h ON h.ts_code = sb.ts_code AND h.user_id = :uid + LEFT JOIN daily_snapshot snap ON snap.ts_code = sb.ts_code + AND snap.trade_date = (SELECT max(trade_date) FROM daily_snapshot) + WHERE sb.list_status = 'L' + AND (:search = '' OR sb.symbol LIKE :psearch OR sb.name LIKE :psearch) + AND (:market = '' OR sb.market = :market) + AND (:industry = '' OR sb.industry = :industry) + AND (:area = '' OR sb.area = :area) + AND (:watched_only = false OR w.id IS NOT NULL) + AND (:held_only = false OR h.id IS NOT NULL) + ORDER BY {order_by} + LIMIT :limit OFFSET :offset + ) + SELECT p.ts_code, p.symbol, p.name, p.industry, p.market, p.watched, p.held, + c.close AS close, prev.close AS prev_close, c.ts AS last_ts, + CASE WHEN c.close IS NOT NULL AND prev.close IS NOT NULL AND prev.close <> 0 + THEN round(((c.close / prev.close - 1) * 100)::numeric, 2) END AS pct_chg, + p.turnover_rate, p.pe_ttm, p.pb, + round((p.total_mv / 10000.0)::numeric, 2) AS total_mv, + round((p.circ_mv / 10000.0)::numeric, 2) AS circ_mv + FROM page p + LEFT JOIN LATERAL ( + SELECT close, ts FROM candles + WHERE symbol = p.symbol AND timeframe = '1d' + ORDER BY ts DESC LIMIT 1 + ) c ON true + LEFT JOIN LATERAL ( + SELECT close FROM candles + WHERE symbol = p.symbol AND timeframe = '1d' AND ts < c.ts + ORDER BY ts DESC LIMIT 1 + ) prev ON c.ts IS NOT NULL +""" + + +def _stocks_sql(sort: str, order: str) -> TextClause: + col = _STOCKS_SORTS.get(sort, _STOCKS_SORTS["symbol"]) + direction = "DESC" if order == "desc" else "ASC" + nulls = " NULLS LAST" if col != "sb.symbol" else "" # 快照缺失/亏损无 PE 的排最后 + return text(_STOCKS_SQL_TMPL.format(order_by=f"{col} {direction}{nulls}")) + +_STOCKS_COUNT_SQL = text(""" + SELECT count(*) FROM stock_basic sb + LEFT JOIN watchlist_items w ON w.ts_code = sb.ts_code AND w.user_id = :uid + LEFT JOIN holding_items h ON h.ts_code = sb.ts_code AND h.user_id = :uid + WHERE sb.list_status = 'L' + AND (:search = '' OR sb.symbol LIKE :psearch OR sb.name LIKE :psearch) + AND (:market = '' OR sb.market = :market) + AND (:industry = '' OR sb.industry = :industry) + AND (:area = '' OR sb.area = :area) + AND (:watched_only = false OR w.id IS NOT NULL) + AND (:held_only = false OR h.id IS NOT NULL) +""") + + +@router.get("/stocks", response_model=StockListResponse) +async def list_stocks( + search: str = "", + market: str = "", + industry: str = "", + area: str = "", + watched_only: bool = False, + held_only: bool = False, + sort: str = "symbol", + order: str = "asc", + limit: int = 100, + offset: int = 0, + session: AsyncSession = Depends(get_session), + user=Depends(require_user), +) -> Response: + """全市场股票列表：stock_basic 基本信息 + candles 最新行情 + daily_snapshot 估值指标 + （换手率/PE-TTM/PB/市值，无快照则这些列为空）。 + watched_only=true 只看自选，held_only=true 只看持仓（各有独立分类入口，列表不再把它们排最前）。 + sort ∈ {symbol,total_mv,circ_mv,pe_ttm,pb,turnover_rate}（白名单，其他值回落 symbol）， + order ∈ asc/desc；快照列排序时缺失值（无快照/亏损无 PE）恒排末尾。 + 缓存：按「用户自选/持仓版本 + 查询参数（含排序）」缓存整页（含 total）；自选/持仓增删即时失效； + 与 preview 同款序列化 JSON 直返（j 前缀），命中跳过 pydantic 校验/序列化。""" + search = search.strip() + sort = sort if sort in _STOCKS_SORTS else "symbol" + order = "desc" if order.lower() == "desc" else "asc" + limit = max(1, min(limit, 500)) + offset = max(0, offset) + key = ( + f"stocksj:u{user.id}" + f":v{await cache.get_version(f'watchlist:{user.id}')}" + f":h{await cache.get_version(f'holding:{user.id}')}" + f":{cache.digest(search, market, industry, area, watched_only, held_only, sort, order, limit, offset)}" + ) + cached = await cached_json_response(key) + if cached is not None: + return cached + params = { + "search": search, + "psearch": f"%{search}%", + "market": market, + "industry": industry, + "area": area, + "watched_only": watched_only, + "held_only": held_only, + "uid": user.id, + "limit": limit, + "offset": offset, + } + total = (await session.execute(_STOCKS_COUNT_SQL, params)).scalar_one() + rows = (await session.execute(_stocks_sql(sort, order), params)).mappings().all() + resp = StockListResponse(total=total, items=[StockListItemOut(**r) for r in rows]) + raw = raw_json(resp) + cache.local_set(key, raw, ttl=min(120, settings.stocks_cache_ttl)) + cache.set_bg(key, raw, ttl=settings.stocks_cache_ttl) + return Response(content=raw, media_type="application/json") + + +@router.get("/stocks/facets", response_model=StockFacetsResponse) +async def stock_facets(session: AsyncSession = Depends(get_session)) -> Response: + """看股页筛选项：行业 / 地域（含数量，按数量降序）。stock_basic 很少变，长缓存。""" + cached = await cached_json_response("facetsj:stocks") + if cached is not None: + return cached + industries = ( + await session.execute(text(""" + SELECT industry AS name, count(*) AS n FROM stock_basic + WHERE list_status = 'L' AND industry IS NOT NULL AND industry <> '' + GROUP BY industry ORDER BY n DESC + """)) + ).mappings().all() + areas = ( + await session.execute(text(""" + SELECT area AS name, count(*) AS n FROM stock_basic + WHERE list_status = 'L' AND area IS NOT NULL AND area <> '' + GROUP BY area ORDER BY n DESC + """)) + ).mappings().all() + resp = StockFacetsResponse( + industries=[FacetItemOut(name=r["name"], count=r["n"]) for r in industries], + areas=[FacetItemOut(name=r["name"], count=r["n"]) for r in areas], + ) + raw = raw_json(resp) + cache.local_set("facetsj:stocks", raw, ttl=min(120, settings.facets_cache_ttl)) + cache.set_bg("facetsj:stocks", raw, ttl=settings.facets_cache_ttl) + return Response(content=raw, media_type="application/json") + + +@router.get("/stocks/{ts_code}/company", response_model=StockCompanyOut) +async def stock_company_info(ts_code: str, session: AsyncSession = Depends(get_session)) -> StockCompanyOut: + """公司简介：库内有新鲜行直返；否则锁内单查 tushare（stock_company）并 upsert（行即缓存， + 30 天新鲜度，无此股写墓碑负缓存）。ETF 前置短路；确认无数据 404；tushare 失败且 + 无旧行可降级时 503（有旧行则在数据层降级返回旧行）。""" + code = ts_code.strip().upper() + if "." not in code: + code = to_ts_code(code) # 防御：兼容 6 位裸代码 + if is_etf_symbol(code): + raise HTTPException(status_code=404, detail="ETF 无公司简介") + try: + row = await company_mod.get_company(session, code) + except Exception: + raise HTTPException(status_code=503, detail="tushare 公司简介拉取失败，请稍后重试") + if row is None: + raise HTTPException(status_code=404, detail=f"无公司信息: {code}") + return StockCompanyOut(**row) + + +@router.get("/stocks/{ts_code}/finance", response_model=StockFinanceOut) +async def stock_finance_info(ts_code: str, session: AsyncSession = Depends(get_session)) -> StockFinanceOut: + """财务数据（近五年，按报告期倒序）：库内新鲜直返；否则锁内拉 tushare 四源 + （fina_indicator/income/balancesheet/cashflow）合并 upsert（7 天新鲜度，无数据写墓碑）。 + ETF 前置短路；确认无数据 404；tushare 四源全失败且无旧行可降级时 503。""" + code = ts_code.strip().upper() + if "." not in code: + code = to_ts_code(code) + if is_etf_symbol(code): + raise HTTPException(status_code=404, detail="ETF 无财务数据") + try: + rows = await finance_mod.get_finance(session, code) + except Exception: + raise HTTPException(status_code=503, detail="tushare 财务数据拉取失败，请稍后重试") + if not rows: + raise HTTPException(status_code=404, detail=f"无财务数据: {code}") + return StockFinanceOut(ts_code=code, records=rows) + + +@router.get("/stocks/{ts_code}/dividends", response_model=StockDividendOut) +async def stock_dividend_info(ts_code: str, session: AsyncSession = Depends(get_session)) -> StockDividendOut: + """分红送股（全历史，按分红年度倒序）：库内新鲜直返；否则锁内拉 tushare dividend + 全量替换（7 天新鲜度，无分红写墓碑，空列表是正常返回）。 + ETF 前置短路；tushare 失败且无旧行可降级时 503。""" + code = ts_code.strip().upper() + if "." not in code: + code = to_ts_code(code) + if is_etf_symbol(code): + raise HTTPException(status_code=404, detail="ETF 无分红数据") + try: + rows = await dividend_mod.get_dividends(session, code) + except Exception: + raise HTTPException(status_code=503, detail="tushare 分红数据拉取失败，请稍后重试") + return StockDividendOut(ts_code=code, records=rows) + + +@router.get("/stocks/{ts_code}/reference/{kind}", response_model=StockReferenceOut) +async def stock_reference_info(ts_code: str, kind: str, session: AsyncSession = Depends(get_session)) -> StockReferenceOut: + """参考数据（11 类，kind 白名单见 reference.REFERENCE_KINDS）：单股单分类 JSON 快照 + 懒加载（7 天新鲜度，无数据写墓碑，空 records 是正常返回）。repurchase 为全市场 + 按月回填的特殊管道：首次触发后台回填近 24 个月（本次可能返回空，稍后再看）。 + ETF 前置短路；未知 kind 404；tushare 失败且无旧行可降级时 503。""" + code = ts_code.strip().upper() + if "." not in code: + code = to_ts_code(code) + if is_etf_symbol(code): + raise HTTPException(status_code=404, detail="ETF 无参考数据") + if kind not in reference_mod.REFERENCE_KINDS: + raise HTTPException(status_code=404, detail=f"未知参考数据分类: {kind}") + try: + rows = await reference_mod.get_reference(session, code, kind) + except Exception: + raise HTTPException(status_code=503, detail="tushare 参考数据拉取失败，请稍后重试") + return StockReferenceOut(ts_code=code, kind=kind, records=rows) diff --git a/backend/app/api/user.py b/backend/app/api/user.py new file mode 100644 index 0000000..f298d2a --- /dev/null +++ b/backend/app/api/user.py @@ -0,0 +1,291 @@ +"""用户数据路由：偏好 / 自选股 / 交割单（个人实盘买卖点）。""" +from __future__ import annotations + +import json + +from fastapi import APIRouter, Depends, File, HTTPException, UploadFile +from sqlalchemy import delete, select, text +from sqlalchemy.ext.asyncio import AsyncSession + +from .. import cache +from ..auth import require_user +from ..db import get_session +from ..models import HoldingItem, StockBasic, UserPreference, UserTrade, WatchlistItem +from ..schemas import ( + HoldingOp, + PreferencesOut, + PreferencesUpdate, + TradesClearResponse, + TradesImportResponse, + UserTradeOut, + WatchlistOp, +) +from ..trades import parse_statement + +router = APIRouter() + + +# ---------- 用户偏好 ---------- +@router.get("/preferences", response_model=PreferencesOut) +async def get_preferences( + session: AsyncSession = Depends(get_session), user=Depends(require_user) +) -> PreferencesOut: + prefs: dict[str, object] = {} + rows = ( + await session.execute(select(UserPreference).where(UserPreference.user_id == user.id)) + ).scalars().all() + for r in rows: + try: + prefs[r.key] = json.loads(r.value_json) + except Exception: # noqa: BLE001 + prefs[r.key] = None + return PreferencesOut(prefs=prefs) + + +@router.put("/preferences", response_model=PreferencesOut) +async def put_preferences( + req: PreferencesUpdate, + session: AsyncSession = Depends(get_session), + user=Depends(require_user), +) -> PreferencesOut: + """部分更新：只覆盖出现的 key；值为 null 表示删除该 key。返回更新后的全量。""" + for key, value in req.prefs.items(): + if not key or len(key) > 64: + continue + if value is None: + await session.execute( + text("DELETE FROM user_preferences WHERE user_id = :u AND key = :k"), + {"u": user.id, "k": key}, + ) + continue + existing = ( + await session.execute( + select(UserPreference).where( + UserPreference.user_id == user.id, UserPreference.key == key + ) + ) + ).scalars().first() + vj = json.dumps(value, ensure_ascii=False) + if existing: + existing.value_json = vj + else: + session.add(UserPreference(user_id=user.id, key=key, value_json=vj)) + await session.commit() + return await get_preferences(session=session, user=user) + + +# ---------- 自选股 ---------- +@router.get("/watchlist", response_model=list[str]) +async def get_watchlist( + session: AsyncSession = Depends(get_session), user=Depends(require_user) +) -> list[str]: + """当前用户自选股 ts_code 列表（加入时间倒序）。""" + rows = ( + await session.execute( + select(WatchlistItem.ts_code) + .where(WatchlistItem.user_id == user.id) + .order_by(WatchlistItem.created_at.desc(), WatchlistItem.id.desc()) + ) + ).scalars().all() + return list(rows) + + +@router.post("/watchlist", response_model=list[str]) +async def add_watchlist( + req: WatchlistOp, + session: AsyncSession = Depends(get_session), + user=Depends(require_user), +) -> list[str]: + exists = ( + await session.execute( + select(WatchlistItem.id).where( + WatchlistItem.user_id == user.id, WatchlistItem.ts_code == req.ts_code + ) + ) + ).scalar_one_or_none() + if exists is None: + session.add(WatchlistItem(user_id=user.id, ts_code=req.ts_code)) + await session.commit() + await cache.bump_version(f"watchlist:{user.id}") # 作废该用户的股票列表缓存 + return await get_watchlist(session=session, user=user) + + +@router.delete("/watchlist/{ts_code}", response_model=list[str]) +async def remove_watchlist( + ts_code: str, + session: AsyncSession = Depends(get_session), + user=Depends(require_user), +) -> list[str]: + await session.execute( + text("DELETE FROM watchlist_items WHERE user_id = :u AND ts_code = :c"), + {"u": user.id, "c": ts_code}, + ) + await session.commit() + await cache.bump_version(f"watchlist:{user.id}") # 作废该用户的股票列表缓存 + return await get_watchlist(session=session, user=user) + + +# ---------- 持仓股 ---------- +@router.get("/holdings", response_model=list[str]) +async def get_holdings( + session: AsyncSession = Depends(get_session), user=Depends(require_user) +) -> list[str]: + """当前用户持仓股 ts_code 列表（加入时间倒序）。""" + rows = ( + await session.execute( + select(HoldingItem.ts_code) + .where(HoldingItem.user_id == user.id) + .order_by(HoldingItem.created_at.desc(), HoldingItem.id.desc()) + ) + ).scalars().all() + return list(rows) + + +@router.post("/holdings", response_model=list[str]) +async def add_holding( + req: HoldingOp, + session: AsyncSession = Depends(get_session), + user=Depends(require_user), +) -> list[str]: + exists = ( + await session.execute( + select(HoldingItem.id).where( + HoldingItem.user_id == user.id, HoldingItem.ts_code == req.ts_code + ) + ) + ).scalar_one_or_none() + if exists is None: + session.add(HoldingItem(user_id=user.id, ts_code=req.ts_code)) + await session.commit() + await cache.bump_version(f"holding:{user.id}") # 作废该用户的股票列表缓存 + return await get_holdings(session=session, user=user) + + +@router.delete("/holdings/{ts_code}", response_model=list[str]) +async def remove_holding( + ts_code: str, + session: AsyncSession = Depends(get_session), + user=Depends(require_user), +) -> list[str]: + await session.execute( + text("DELETE FROM holding_items WHERE user_id = :u AND ts_code = :c"), + {"u": user.id, "c": ts_code}, + ) + await session.commit() + await cache.bump_version(f"holding:{user.id}") # 作废该用户的股票列表缓存 + return await get_holdings(session=session, user=user) + + +# ---------- 交割单（个人实盘买卖点） ---------- +def _trade_out(r: UserTrade) -> UserTradeOut: + return UserTradeOut( + id=r.id, ts_code=r.ts_code, name=r.name, trade_date=r.trade_date, + direction=r.direction, price=r.price, qty=r.qty, amount=r.amount, fee=r.fee, + ) + + +@router.get("/trades", response_model=list[UserTradeOut]) +async def list_trades( + ts_code: str | None = None, + session: AsyncSession = Depends(get_session), + user=Depends(require_user), +) -> list[UserTradeOut]: + """当前用户导入的实盘成交（可选 ts_code 过滤，按日期升序；K线买卖点数据源）。""" + q = ( + select(UserTrade) + .where(UserTrade.user_id == user.id) + .order_by(UserTrade.trade_date, UserTrade.id) + ) + if ts_code: + q = q.where(UserTrade.ts_code == ts_code) + rows = (await session.execute(q)).scalars().all() + return [_trade_out(r) for r in rows] + + +@router.post("/trades/import", response_model=TradesImportResponse) +async def import_trades( + file: UploadFile = File(...), + session: AsyncSession = Depends(get_session), + user=Depends(require_user), +) -> TradesImportResponse: + """上传券商交割单（CSV/Excel/HTML 表格均可，自动识别列名），解析出买卖成交入库。 + + 同一笔成交（同日同股同向同价同量）重复上传会跳过，重复导出幂等。 + """ + data = await file.read() + if not data: + raise HTTPException(status_code=422, detail="文件是空的") + if len(data) > 20 * 1024 * 1024: + raise HTTPException(status_code=413, detail="文件超过 20MB，请分时间段导出") + + parsed = parse_statement(data, file.filename or "") + + # 无证券代码列的导出（招商式）：按证券名称反查 stock_basic 补 ts_code；同名多码或查不到则弃行 + unnamed = {t.name for t in parsed.trades if not t.ts_code and t.name} + if unnamed: + name_map: dict[str, str] = {} + for ts_code, name in (await session.execute( + select(StockBasic.ts_code, StockBasic.name).where(StockBasic.name.in_(unnamed)) + )).all(): + name_map[name] = "" if name in name_map else ts_code + for t in parsed.trades: + if not t.ts_code and t.name: + tc = name_map.get(t.name, "") + if tc: + t.ts_code, t.code = tc, tc.split(".")[0] + else: + parsed.skipped_bad.append(f"{t.trade_date} {t.name} 名称无法唯一对应代码，未入库") + + def _key(t) -> tuple: + return (t.trade_date, t.ts_code, t.direction, None if t.price is None else round(t.price, 4), round(t.qty, 4)) + + # Python 侧去重兜底（唯一约束对 NULL price 不生效） + existing = { + (r.trade_date, r.ts_code, r.direction, None if r.price is None else round(r.price, 4), round(r.qty, 4)) + for r in ( + await session.execute( + select(UserTrade.trade_date, UserTrade.ts_code, UserTrade.direction, UserTrade.price, UserTrade.qty) + .where(UserTrade.user_id == user.id, UserTrade.ts_code.in_({t.ts_code for t in parsed.trades})) + ) + ).all() + } + inserted: list[UserTrade] = [] + seen: set[tuple] = set() + skipped_dup = 0 + for t in parsed.trades: + if not t.ts_code: + continue # 名称反查失败的行，已在 bad 里说明 + k = _key(t) + if k in existing or k in seen: + skipped_dup += 1 + continue + seen.add(k) + inserted.append(UserTrade( + user_id=user.id, ts_code=t.ts_code, code=t.code, name=t.name or None, + trade_date=t.trade_date, direction=t.direction, price=t.price, + qty=t.qty, amount=t.amount, fee=t.fee, + raw_json=json.dumps(t.raw, ensure_ascii=False, default=str), + )) + if inserted: + session.add_all(inserted) + await session.commit() + + return TradesImportResponse( + inserted=len(inserted), + skipped_dup=skipped_dup, + skipped_other=parsed.skipped_other, + stocks=len({t.ts_code for t in parsed.trades}), + bad=parsed.skipped_bad[:5], + sample=[_trade_out(r) for r in inserted[:5]], + ) + + +@router.delete("/trades", response_model=TradesClearResponse) +async def clear_trades( + session: AsyncSession = Depends(get_session), + user=Depends(require_user), +) -> TradesClearResponse: + """清空当前用户导入的全部成交（重新导入前用）。""" + res = await session.execute(delete(UserTrade).where(UserTrade.user_id == user.id)) + await session.commit() + return TradesClearResponse(deleted=res.rowcount or 0) diff --git a/backend/app/data/limit_board.py b/backend/app/data/limit_board.py new file mode 100644 index 0000000..d404b6b --- /dev/null +++ b/backend/app/data/limit_board.py @@ -0,0 +1,233 @@ +"""首页打板专题（tushare 同花顺版：limit_list_ths / limit_step / limit_cpt_list）。 + +整包 SWR（仿 index_global 列表层）：进程内 state 新鲜直返 -> Redis 回填 -> +有旧值先返 + 后台刷新 -> 冷启动同步拉。盘中（交易日 09:15-15:30）数据源即有 +当日快照（实测 quicksync 镜像盘中可取当日）-> fresh TTL 压到 5 分钟；其余时段 4 小时。 + +镜像坑（见 reference.py 注释）：不传 fields；limit_list_ths 必须显式传 trade_date +（缺省返回多日混包且 4000 行封顶）；涨停/连扳池才有 tag/status/lu_desc/封单额， +炸板池只有价格与打开次数，跌停池几乎只有价格——行模型统一、字段可选。 +""" +from __future__ import annotations + +import asyncio +import math +import time +from datetime import date, datetime, timedelta + +from .. import cache +from ..config import settings +from .sync_utils import call_retry, f_clean, get_pro_lazy, s_clean + +_BOARD_KEY = "limit_board:daily:v1" # Redis 整包缓存键（v1 起版） +_REDIS_TTL = 3600 # 进程重启后的回填来源 +_INTRADAY_TTL = 300.0 # 交易时段内的 fresh TTL（5 分钟准实时） +_MAX_DATE_BACKTRACK = 5 # trade_date 定位回退天数（覆盖节假日/盘前） +_BLOCKS_OUT = 12 # 最强板块输出条数 + + +class LimitBoardError(RuntimeError): + """三池全部拉不到（token/网络故障）——接口层转 503。""" + + +def _yi(v) -> float | None: + """元 -> 亿元（2 位小数）。""" + f = f_clean(v) + return None if f is None else round(f / 1e8, 2) + + +def _fetch_pool_sync(pro, trade_date: str, limit_type: str): + time.sleep(settings.screener_sync_interval) + return call_retry(pro.limit_list_ths, trade_date=trade_date, limit_type=limit_type) + + +def _pool_rows(df, mode: str) -> list[dict]: + """行裁剪 + 单位换算。mode: up / broken / down。""" + if df is None or df.empty: + return [] + rows: list[dict] = [] + for _, r in df.iterrows(): + row = { + "ts_code": s_clean(r.get("ts_code")), + "name": s_clean(r.get("name")), + "price": f_clean(r.get("price")), + "pct_chg": f_clean(r.get("pct_chg")), + } + if not row["ts_code"]: + continue + if mode == "up": + row.update({ + "tag": s_clean(r.get("tag")), + "status": s_clean(r.get("status")), + "lu_desc": s_clean(r.get("lu_desc")), + "open_num": f_clean(r.get("open_num")), + "limit_amount_yi": _yi(r.get("limit_amount")), # 封单额（亿） + "turnover_yi": _yi(r.get("turnover")), # 成交额（亿） + "first_lu_time": s_clean(r.get("first_lu_time")), + "limit_up_suc_rate": f_clean(r.get("limit_up_suc_rate")), + }) + elif mode == "broken": + row.update({ + "open_num": f_clean(r.get("open_num")), + "first_lu_time": s_clean(r.get("first_lu_time")), + "last_lu_time": s_clean(r.get("last_lu_time")), + }) + rows.append(row) + if mode == "up": + # 封单额降序（打板看封单强度）；封单额缺失（镜像个别行）沉底 + rows.sort(key=lambda x: (x.get("limit_amount_yi") is None, -(x.get("limit_amount_yi") or 0))) + return rows + + +def _fetch_board_sync() -> dict: + """定位交易日并拉三池 + 天梯 + 最强板块（同步网络 IO，需在 to_thread 里跑）。""" + pro = get_pro_lazy() + errors: list[str] = [] + + # trade_date 定位：今日起逐日回退，取第一个涨停池非空的日期 + # （盘前/节假日当日为空；tushare 错误直接抛——定位失败无意义继续） + trade_date: str | None = None + up_rows: list[dict] = [] + for i in range(_MAX_DATE_BACKTRACK): + d = (date.today() - timedelta(days=i)).strftime("%Y%m%d") + df = _fetch_pool_sync(pro, d, "涨停池") + if df is not None and not df.empty: + trade_date = d + up_rows = _pool_rows(df, "up") + break + if trade_date is None: + raise LimitBoardError(f"近 {_MAX_DATE_BACKTRACK} 天均无涨停池数据（节假日或数据源故障）") + + broken_rows: list[dict] = [] + try: + broken_rows = _pool_rows(_fetch_pool_sync(pro, trade_date, "炸板池"), "broken") + except Exception as e: # noqa: BLE001 —— 单池失败不拖垮整包 + errors.append(f"炸板池: {str(e)[:60]}") + + down_rows: list[dict] = [] + try: + down_rows = _pool_rows(_fetch_pool_sync(pro, trade_date, "跌停池"), "down") + except Exception as e: # noqa: BLE001 + errors.append(f"跌停池: {str(e)[:60]}") + + ladder: list[dict] = [] + try: + time.sleep(settings.screener_sync_interval) + step = call_retry(pro.limit_step, trade_date=trade_date) + if step is not None and not step.empty: + for _, r in step.iterrows(): + code = s_clean(r.get("ts_code")) + n = f_clean(r.get("nums")) + if code and n: + ladder.append({"ts_code": code, "name": s_clean(r.get("name")), "nums": int(n)}) + ladder.sort(key=lambda x: -x["nums"]) + except Exception as e: # noqa: BLE001 + errors.append(f"连板天梯: {str(e)[:60]}") + + blocks: list[dict] = [] + try: + time.sleep(settings.screener_sync_interval) + cpt = call_retry(pro.limit_cpt_list, trade_date=trade_date) + if cpt is not None and not cpt.empty: + for _, r in cpt.head(_BLOCKS_OUT).iterrows(): + blocks.append({ + "name": s_clean(r.get("name")), + "days": f_clean(r.get("days")), + "up_stat": s_clean(r.get("up_stat")), + "cons_nums": f_clean(r.get("cons_nums")), + "up_nums": f_clean(r.get("up_nums")), + "pct_chg": f_clean(r.get("pct_chg")), + }) + except Exception as e: # noqa: BLE001 + errors.append(f"最强板块: {str(e)[:60]}") + + # 连板分布（limit_step 只含 2 板及以上；1 板 = 涨停池 tag 首板数） + dist: dict[int, int] = {} + for x in ladder: + dist[x["nums"]] = dist.get(x["nums"], 0) + 1 + summary = { + "up_count": len(up_rows), + "broken_count": len(broken_rows), + "down_count": len(down_rows), + "first_board_count": sum(1 for x in up_rows if x.get("tag") == "首板"), + "max_ladder": ladder[0] if ladder else None, + "ladder_dist": [{"nums": k, "count": v} for k, v in sorted(dist.items())], + } + return { + "trade_date": f"{trade_date[:4]}-{trade_date[4:6]}-{trade_date[6:]}", + "summary": summary, + "up": up_rows, + "broken": broken_rows, + "down": down_rows, + "ladder": ladder, + "blocks": blocks, + "errors": errors, + } + + +# ---------- 整包 SWR（进程内 -> Redis 回填 -> 旧值先返 + 后台刷新 -> 冷启动同步拉） ---------- + +_state: dict = {"payload": None} +_refreshing = False +_refresh_error: str | None = None +_bg_tasks: set[asyncio.Task] = set() + + +def _fresh_ttl(is_trading_day: bool | None) -> float: + """交易时段 5 分钟（镜像盘中即有当日快照）；其余 4 小时（盘后数据不变）。""" + if is_trading_day is None: + is_trading_day = datetime.now().weekday() < 5 # 判定失败回退 weekday 启发式 + if is_trading_day: + now = datetime.now() + t = now.hour * 60 + now.minute + if 9 * 60 + 15 <= t <= 15 * 60 + 30: + return _INTRADAY_TTL + return float(settings.market_eod_fresh_ttl) + + +async def _refresh() -> dict: + data = await asyncio.to_thread(_fetch_board_sync) + payload = {**data, "updated_at": datetime.now().isoformat(), "fetched_ts": time.time()} + _state["payload"] = payload + await cache.cache_set(_BOARD_KEY, payload, ttl=_REDIS_TTL) + return payload + + +async def _refresh_wrapped() -> None: + global _refresh_error, _refreshing + try: + await _refresh() + _refresh_error = None + except Exception as e: # noqa: BLE001 —— 后台刷新失败静默记错，下次并入 errors + _refresh_error = f"打板专题后台刷新: {str(e)[:60]}" + finally: + _refreshing = False + + +def _spawn_refresh() -> None: + global _refreshing + if _refreshing: + return + _refreshing = True + task = asyncio.create_task(_refresh_wrapped()) + _bg_tasks.add(task) + task.add_done_callback(_bg_tasks.discard) + + +async def fetch_limit_board(is_trading_day: bool | None) -> dict: + """打板专题整包读取（SWR）。冷启动同步拉（5 次调用约 2-4s）；此后盘中 5 分钟/盘后 4 小时。""" + ttl = _fresh_ttl(is_trading_day) + p = _state["payload"] + if p is not None and time.time() - p["fetched_ts"] < ttl: + return p + if p is None: + cached = await cache.cache_get(_BOARD_KEY) + if cached: + p = cached + _state["payload"] = p + if p is not None: + _spawn_refresh() + if _refresh_error and not p.get("errors"): + p = {**p, "errors": [_refresh_error]} + return p + return await _refresh() diff --git a/backend/app/data/ths_board.py b/backend/app/data/ths_board.py new file mode 100644 index 0000000..ccc2b94 --- /dev/null +++ b/backend/app/data/ths_board.py @@ -0,0 +1,242 @@ +"""同花顺概念/行业板块（ths_index 列表 + ths_daily 行情快照 + ths_member 成分）。 + +缓存分层（数据特性决定）： +- 板块列表：一天不变 -> 直缓存（进程内 -> Redis 24h） +- 行情快照：ths_daily 盘中即有当日（实测镜像），全市场一日 1877 行单次拿全 + -> 整包 SWR（同 limit_board：盘中 5 分钟 / 盘后 4 小时，trade_date 回退定位） +- 成分：每板块懒加载直缓存 24h；成分股行情 enrich（candles 最新+前收 LATERAL） + 每次请求现算，不进缓存 + +type 代码：N 概念 / I 行业 / TH 主题 / S 特色 / R 地域 / BB 宽基 / ST 风格。 +""" +from __future__ import annotations + +import asyncio +import time +from datetime import date, datetime, timedelta + +from sqlalchemy import text +from sqlalchemy.ext.asyncio import AsyncSession + +from .. import cache +from ..config import settings +from .sync_utils import call_retry, f_clean, get_pro_lazy, s_clean + +_LIST_KEY = "ths_boards:list:v1" +_DAILY_KEY = "ths_boards:daily:v1" +_REDIS_TTL = 3600 +_LIST_TTL = 86400 +_INTRADAY_TTL = 300.0 +_MAX_DATE_BACKTRACK = 5 + + +class ThsBoardError(RuntimeError): + """列表/行情全部拉不到 —— 接口层转 503。""" + + +# ---------- 板块列表（直缓存：进程内 -> Redis 24h -> 拉取） ---------- + +_list_cache: list[dict] | None = None + + +def _fetch_list_sync() -> list[dict]: + time.sleep(settings.screener_sync_interval) + df = call_retry(get_pro_lazy().ths_index, exchange="A") + if df is None or df.empty: + raise ThsBoardError("ths_index 板块列表为空") + rows: list[dict] = [] + for _, r in df.iterrows(): + code = s_clean(r.get("ts_code")) + if not code: + continue + rows.append({ + "ts_code": code, + "name": s_clean(r.get("name")), + "type": s_clean(r.get("type")), + "count": f_clean(r.get("count")), + "list_date": s_clean(r.get("list_date")), + }) + return rows + + +async def get_board_list() -> list[dict]: + global _list_cache + if _list_cache is not None: + return _list_cache + cached = await cache.cache_get(_LIST_KEY) + if isinstance(cached, list) and cached: + _list_cache = cached + return cached + rows = await asyncio.to_thread(_fetch_list_sync) + _list_cache = rows + await cache.cache_set(_LIST_KEY, rows, ttl=_LIST_TTL) + return rows + + +# ---------- 行情快照（整包 SWR，同 limit_board.py 三件套） ---------- + +_daily_state: dict = {"payload": None} +_daily_refreshing = False +_daily_refresh_error: str | None = None +_bg_tasks: set[asyncio.Task] = set() + + +def _fresh_ttl(is_trading_day: bool | None) -> float: + """交易时段 5 分钟（镜像盘中即有当日快照）；其余 4 小时。""" + if is_trading_day is None: + is_trading_day = datetime.now().weekday() < 5 + if is_trading_day: + now = datetime.now() + t = now.hour * 60 + now.minute + if 9 * 60 + 15 <= t <= 15 * 60 + 30: + return _INTRADAY_TTL + return float(settings.market_eod_fresh_ttl) + + +def _fetch_daily_sync() -> dict: + pro = get_pro_lazy() + for i in range(_MAX_DATE_BACKTRACK): + d = (date.today() - timedelta(days=i)).strftime("%Y%m%d") + time.sleep(settings.screener_sync_interval) + df = call_retry(pro.ths_daily, trade_date=d) + if df is None or df.empty: + continue + quotes: dict[str, dict] = {} + for _, r in df.iterrows(): + code = s_clean(r.get("ts_code")) + if code: + quotes[code] = { + "close": f_clean(r.get("close")), + "pct_change": f_clean(r.get("pct_change")), + "vol": f_clean(r.get("vol")), + "turnover_rate": f_clean(r.get("turnover_rate")), + } + return {"trade_date": f"{d[:4]}-{d[4:6]}-{d[6:]}", "quotes": quotes} + raise ThsBoardError(f"近 {_MAX_DATE_BACKTRACK} 天均无 ths_daily 板块行情") + + +async def _refresh_daily() -> dict: + data = await asyncio.to_thread(_fetch_daily_sync) + payload = {**data, "updated_at": datetime.now().isoformat(), "fetched_ts": time.time()} + _daily_state["payload"] = payload + await cache.cache_set(_DAILY_KEY, payload, ttl=_REDIS_TTL) + return payload + + +async def _refresh_daily_wrapped() -> None: + global _daily_refresh_error, _daily_refreshing + try: + await _refresh_daily() + _daily_refresh_error = None + except Exception as e: # noqa: BLE001 + _daily_refresh_error = f"板块行情后台刷新: {str(e)[:60]}" + finally: + _daily_refreshing = False + + +def _spawn_daily_refresh() -> None: + global _daily_refreshing + if _daily_refreshing: + return + _daily_refreshing = True + task = asyncio.create_task(_refresh_daily_wrapped()) + _bg_tasks.add(task) + task.add_done_callback(_bg_tasks.discard) + + +async def _get_daily(is_trading_day: bool | None) -> dict: + ttl = _fresh_ttl(is_trading_day) + p = _daily_state["payload"] + if p is not None and time.time() - p["fetched_ts"] < ttl: + return p + if p is None: + cached = await cache.cache_get(_DAILY_KEY) + if cached: + p = cached + _daily_state["payload"] = p + if p is not None: + _spawn_daily_refresh() + return p + return await _refresh_daily() + + +async def fetch_boards(is_trading_day: bool | None) -> dict: + """列表 + 当日行情合并（行情缺失的板块价格为 null）。""" + boards, daily = await asyncio.gather(get_board_list(), _get_daily(is_trading_day)) + quotes: dict = daily.get("quotes", {}) + merged = [{**b, **quotes.get(b["ts_code"], {})} for b in boards] + errors = [_daily_refresh_error] if _daily_refresh_error else [] + return { + "trade_date": daily.get("trade_date"), + "updated_at": daily.get("updated_at"), + "boards": merged, + "errors": errors, + } + + +# ---------- 成分（每板块懒加载直缓存 24h + 行情 enrich 现算） ---------- + +_member_cache: dict[str, list[dict]] = {} + + +def _fetch_members_sync(board_code: str) -> list[dict]: + time.sleep(settings.screener_sync_interval) + df = call_retry(get_pro_lazy().ths_member, ts_code=board_code) + if df is None or df.empty: + return [] + rows: list[dict] = [] + for _, r in df.iterrows(): + code = s_clean(r.get("con_code")) + if code: + rows.append({"con_code": code, "con_name": s_clean(r.get("con_name"))}) + rows.sort(key=lambda x: x["con_code"]) + return rows + + +async def get_raw_members(board_code: str) -> list[dict]: + hit = _member_cache.get(board_code) + if hit is not None: + return hit + key = f"ths_members:{board_code}" + cached = await cache.cache_get(key) + if isinstance(cached, list): + _member_cache[board_code] = cached + return cached + rows = await asyncio.to_thread(_fetch_members_sync, board_code) + _member_cache[board_code] = rows + await cache.cache_set(key, rows, ttl=_LIST_TTL) + return rows + + +# 成分股行情：candles 最新价 + 前收算涨跌幅（北交所等无底座数据的为 NULL） +_MEMBERS_ENRICH_SQL = text(""" + SELECT m.code AS con_code, + c.close AS close, + CASE WHEN c.close IS NOT NULL AND prev.close IS NOT NULL AND prev.close <> 0 + THEN round(((c.close / prev.close - 1) * 100)::numeric, 2) END AS pct_chg + FROM (SELECT unnest(CAST(:codes AS text[])) AS code) m + LEFT JOIN LATERAL ( + SELECT close, ts FROM candles + WHERE symbol = split_part(m.code, '.', 1) AND timeframe = '1d' + ORDER BY ts DESC LIMIT 1 + ) c ON true + LEFT JOIN LATERAL ( + SELECT close FROM candles + WHERE symbol = split_part(m.code, '.', 1) AND timeframe = '1d' AND ts < c.ts + ORDER BY ts DESC LIMIT 1 + ) prev ON c.ts IS NOT NULL +""") + + +async def get_members(session: AsyncSession, board_code: str) -> list[dict]: + """成分 + 现价/涨跌幅（行情不缓存，每请求现算）。""" + members = await get_raw_members(board_code) + if not members: + return [] + rows = (await session.execute(_MEMBERS_ENRICH_SQL, { + "codes": [m["con_code"] for m in members], + })).mappings().all() + quote = {r["con_code"]: {"close": float(r["close"]) if r["close"] is not None else None, + "pct_chg": float(r["pct_chg"]) if r["pct_chg"] is not None else None} + for r in rows} + return [{**m, **quote.get(m["con_code"], {"close": None, "pct_chg": None})} for m in members] diff --git a/backend/app/models.py b/backend/app/models.py index 5a92066..1cf4528 100644 --- a/backend/app/models.py +++ b/backend/app/models.py @@ -312,6 +312,20 @@ class WatchlistItem(Base): ) +class HoldingItem(Base): + """持仓股（手动标记进「持仓」分类的股票）。""" + __tablename__ = "holding_items" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + user_id: Mapped[int] = mapped_column(BigInteger, ForeignKey("users.id", ondelete="CASCADE"), index=True) + ts_code: Mapped[str] = mapped_column(String(12), index=True) + created_at: Mapped[datetime] = mapped_column(DateTime, default=_utcnow) + + __table_args__ = ( + UniqueConstraint("user_id", "ts_code", name="uq_holding_user_code"), + ) + + class UserTrade(Base): """交割单导入的实盘成交流水（K线买卖点的数据源，价格为券商成交原始价、不复权）。""" __tablename__ = "user_trades" diff --git a/backend/app/scheduler.py b/backend/app/scheduler.py new file mode 100644 index 0000000..395b43a --- /dev/null +++ b/backend/app/scheduler.py @@ -0,0 +1,114 @@ +"""夜间定时任务：收盘后自动全市场同步（A股 + ETF）+ 过期会话清理。 + +进程内 asyncio 循环（单进程部署假设）。start_sync 均幂等（已在跑直接返回）， +即使多 worker / 手动触发与定时撞车也不会重复跑。关闭：NIGHTLY_SYNC_ENABLED=false。 + +补跑语义：进程启动时若已过触发点、今天是交易日、且当日 candles 尚未落库 +（例如定时点机器没开机），立即补跑一次，不让数据断档等到第二天。 +""" +from __future__ import annotations + +import asyncio +import logging +from datetime import datetime, timedelta + +from sqlalchemy import delete, func, or_, select + +from .auth import utcnow +from .config import settings +from .models import AuthSession, Candle, TradeCalendar + +log = logging.getLogger(__name__) + +_TRIGGER_MINUTE = 5 # 触发点 = nightly_sync_hour:05（避开整点拥挤，纯本地任务习惯） + + +def _seconds_until_next_run() -> float: + now = datetime.now() + target = now.replace(hour=settings.nightly_sync_hour, minute=_TRIGGER_MINUTE, + second=0, microsecond=0) + if target <= now: + target += timedelta(days=1) + return (target - now).total_seconds() + + +async def cleanup_sessions() -> int: + """删除过期 / 吊销超 7 天的会话行（登录路径只清本人，长跑进程需要兜底）。""" + from .db import async_session # 延迟导入避免循环 + + now = utcnow() + async with async_session() as s: + res = await s.execute(delete(AuthSession).where(or_( + AuthSession.expires_at < now, + AuthSession.revoked_at.is_not(None) & (AuthSession.revoked_at < now - timedelta(days=7)), + ))) + await s.commit() + return res.rowcount or 0 + + +async def _nightly_routine() -> None: + """当日例行：A 股全市场同步 -> ETF 同步（先后跑，避免两路 tushare 控频挤兑）-> 会话清理。""" + from .data import etf_sync + from .db import async_session + from .screener import market_sync + + log.info("夜间任务开始：全市场同步窗口 %d 个交易日", settings.screener_market_days) + try: + async with async_session() as session: + await market_sync.start_sync(session, days=settings.screener_market_days, force=False) + # 等日线同步收尾再触发 ETF（轮询模块内状态；start_sync 是即发即忘的） + while market_sync._sync_state["running"]: + await asyncio.sleep(30) + except Exception: # noqa: BLE001 + log.exception("夜间 A 股同步触发失败") + try: + await etf_sync.start_sync(full=False) + except Exception: # noqa: BLE001 + log.exception("夜间 ETF 同步触发失败") + try: + n = await cleanup_sessions() + if n: + log.info("夜间会话清理：%d 行", n) + except Exception: # noqa: BLE001 + log.exception("夜间会话清理失败") + log.info("夜间任务结束") + + +async def _should_run_on_startup() -> bool: + """启动补跑判定：已过触发点 + 今天是交易日 + 当日 candles 还没落库。""" + now = datetime.now() + if now.hour * 60 + now.minute < settings.nightly_sync_hour * 60 + _TRIGGER_MINUTE: + return False + today8 = now.strftime("%Y%m%d") + from .db import async_session + + async with async_session() as s: + is_trade_day = bool(await s.scalar( + select(TradeCalendar.id).where(TradeCalendar.trade_date == today8).limit(1))) + if not is_trade_day: + return False + # 当日未收盘/数据未生成时 max(ts) < 今日，同步会拉到空——那正是要补跑的信号 + latest = await s.scalar(select(func.max(Candle.ts)).where(Candle.timeframe == "1d")) + return latest is None or latest.strftime("%Y%m%d") < today8 + + +async def run_nightly_loop() -> None: + """每日触发点跑一次；启动时满足补跑条件先补跑。由 main.lifespan 拉起。""" + if not settings.nightly_sync_enabled: + log.info("夜间自动同步未启用（NIGHTLY_SYNC_ENABLED=false）") + return + try: + if await _should_run_on_startup(): + log.info("启动补跑：已过 %02d:%02d 且当日数据未落库", settings.nightly_sync_hour, _TRIGGER_MINUTE) + await _nightly_routine() + except Exception: # noqa: BLE001 + log.exception("启动补跑判定失败（跳过，等待每日定时点）") + while True: + try: + await asyncio.sleep(_seconds_until_next_run()) + await _nightly_routine() + except asyncio.CancelledError: + raise + except Exception: # noqa: BLE001 —— 循环体不能死 + log.exception("夜间任务循环异常，60s 后继续") + await asyncio.sleep(60) diff --git a/backend/app/schemas.py b/backend/app/schemas.py index 4208080..010b54c 100644 --- a/backend/app/schemas.py +++ b/backend/app/schemas.py @@ -474,6 +474,7 @@ class StockListItemOut(BaseModel): total_mv: float | None = None # 总市值（亿元） circ_mv: float | None = None # 流通市值（亿元） watched: bool = False # 是否自选（当前用户） + held: bool = False # 是否持仓（当前用户） class StockListResponse(BaseModel): @@ -544,6 +545,10 @@ class WatchlistOp(BaseModel): ts_code: str = Field(min_length=6, max_length=12) +class HoldingOp(BaseModel): + ts_code: str = Field(min_length=6, max_length=12) + + class ScreenerQueryOut(BaseModel): id: int text: str diff --git a/backend/conftest.py b/backend/conftest.py new file mode 100644 index 0000000..dfb0a33 --- /dev/null +++ b/backend/conftest.py @@ -0,0 +1 @@ +"""pytest 根 conftest：其存在让 pytest 把 backend/ 加入 sys.path，tests 可直接 import app.*。""" diff --git a/backend/restart_backend.cmd b/backend/restart_backend.cmd new file mode 100644 index 0000000..b29eb0d --- /dev/null +++ b/backend/restart_backend.cmd @@ -0,0 +1,22 @@ +@echo off +rem Restart backend: kill old process tree on port 8000, then start uvicorn with --reload. +rem Log goes to repo root backend_run.log. Usage: cmd /c backend\restart_backend.cmd +rem NOTE: keep this file ASCII only (cmd.exe reads it as GBK, UTF-8 Chinese breaks parsing). +cd /d "%~dp0" + +rem uvicorn --reload spawns launcher -> reloader -> worker, and the reloader holds the :8000 +rem socket. Killing the listener with /T tears down its worker child too, and the launcher +rem exits on its own once its child is gone -- so no orphan survives. (Git Bash: restart_backend.sh) +for /f "tokens=5" %%a in ('netstat -ano ^| findstr /r /c:":8000 .*LISTENING"') do ( + echo kill old PID %%a + taskkill /PID %%a /T /F >nul 2>&1 +) +rem sleep ~2s (GNU timeout from Git Bash PATH shadows cmd timeout, use ping instead) +ping -n 3 127.0.0.1 >nul + +rem Clear SSLKEYLOGFILE (has U+202A control chars that crash asyncpg, same as env -u) +set "SSLKEYLOGFILE=" + +rem redirect must live inside cmd /c - a redirect on the start line is not inherited by the child +start "stock-backend" /min cmd /c ".venv\Scripts\python.exe -m uvicorn app.main:app --reload --port 8000 > ..\backend_run.log 2>&1" +echo backend restarting, log: backend_run.log diff --git a/backend/restart_backend.sh b/backend/restart_backend.sh new file mode 100644 index 0000000..fca27a4 --- /dev/null +++ b/backend/restart_backend.sh @@ -0,0 +1,36 @@ +#!/usr/bin/env bash +# Restart the backend: kill the whole uvicorn process tree on port 8000, then start --reload. +# Log goes to repo root backend_run.log. +# Usage (Git Bash, from repo root): bash backend/restart_backend.sh +set -u + +# Always run relative to this script, so it works from any cwd. +cd "$(dirname "$0")" || exit 1 + +# 1) Kill every python process running THIS backend. uvicorn --reload on Windows +# spawns three processes (top launcher -> reloader -> worker); killing only the +# port listener leaves an orphan reloader that can later rebind :8000 and fight +# the new instance. Match the command line to catch launcher + reloader, then +# taskkill /T tears down each of their descendant trees (the worker included). +pids=$(wmic process where "name='python.exe' and commandline like '%uvicorn app.main%'" \ + get ProcessId 2>/dev/null \ + | tr -d '\r' \ + | grep -Eo '[0-9]+' \ + | sort -u) + +if [ -n "$pids" ]; then + for pid in $pids; do + echo "kill old backend PID $pid" + # //PID double slash stops Git Bash from turning /PID into a path. + taskkill //PID "$pid" //T //F >/dev/null 2>&1 + done + sleep 2 +fi + +# 2) Clear SSLKEYLOGFILE: the user env var's value has a leading U+202A control +# char that makes asyncpg throw OSError [Errno 22] on connect. +unset SSLKEYLOGFILE + +# 3) Start, redirecting to repo root backend_run.log (--reload for hot reload). +echo "starting uvicorn, log: ../backend_run.log" +exec .venv/Scripts/python.exe -m uvicorn app.main:app --reload --port 8000 > ../backend_run.log 2>&1 diff --git a/backend/tests/test_adjust_bars.py b/backend/tests/test_adjust_bars.py new file mode 100644 index 0000000..9a37b64 --- /dev/null +++ b/backend/tests/test_adjust_bars.py @@ -0,0 +1,65 @@ +"""复权换算 adjust_bars：bfq/qfq/hfq 乘数、阶梯因子向前沿用、名义量不缩放。""" +from __future__ import annotations + +from datetime import datetime + +import pytest + +from app.api._deps import adjust_bars +from app.domain import Bar + + +def _bars(*rows) -> list[Bar]: + # (date, open, close) + return [Bar(ts=datetime(y, m, d), open=o, high=o, low=o, close=c, volume=100.0) + for (y, m, d), o, c in rows] + + +def _factors(*pairs): + return [(datetime(y, m, d), f) for (y, m, d), f in pairs] + + +def test_bfq_to_bfq_identity(): + bars = _bars(((2024, 1, 2), 10.0, 11.0)) + out = adjust_bars(bars, _factors(((2024, 1, 2), 2.0)), "bfq", "bfq") + assert out[0].close == 11.0 + + +def test_qfq_normalizes_by_latest_factor(): + # 因子 1.0 -> 2.0（中途除权）：qfq = f(t)/f(latest) + bars = _bars( + ((2024, 1, 2), 10.0, 10.0), # f=1.0 + ((2024, 6, 3), 5.0, 5.0), # f=2.0（除权日，价格腰斩） + ) + factors = _factors(((2024, 1, 2), 1.0), ((2024, 6, 3), 2.0)) + out = adjust_bars(bars, factors, "bfq", "qfq") + # 除权前按 1.0/2.0 缩放 -> 5.0；除权后 2.0/2.0 -> 原价 + assert out[0].close == pytest.approx(5.0) + assert out[1].close == pytest.approx(5.0) + + +def test_hfq_scales_by_factor(): + bars = _bars(((2024, 1, 2), 10.0, 10.0), ((2024, 6, 3), 5.0, 5.0)) + factors = _factors(((2024, 1, 2), 1.0), ((2024, 6, 3), 2.0)) + out = adjust_bars(bars, factors, "bfq", "hfq") + assert out[0].close == pytest.approx(10.0) # f=1 + assert out[1].close == pytest.approx(10.0) # 5 * 2 + + +def test_factor_step_forward_fill(): + """因子是阶梯函数：变化点之间的日期向前沿用最近因子。""" + bars = _bars(((2024, 2, 1), 8.0, 8.0)) # 在 1/2 与 6/3 之间 -> 沿用 1.0 + factors = _factors(((2024, 1, 2), 1.0), ((2024, 6, 3), 2.0)) + out = adjust_bars(bars, factors, "bfq", "hfq") + assert out[0].close == pytest.approx(8.0) # 8 * 1.0 + + +def test_nominal_columns_not_scaled(): + """成交额/换手率是名义量，不随复权缩放。""" + bars = [Bar(ts=datetime(2024, 1, 2), open=10, high=10, low=10, close=10, + volume=100.0, amount=1_000_000.0, turnover=1.5)] + out = adjust_bars(bars, _factors(((2024, 1, 2), 4.0)), "bfq", "hfq") + assert out[0].close == pytest.approx(40.0) + assert out[0].amount == 1_000_000.0 + assert out[0].turnover == 1.5 + assert out[0].volume == 100.0 diff --git a/backend/tests/test_auth_rate_limit.py b/backend/tests/test_auth_rate_limit.py new file mode 100644 index 0000000..f4c313a --- /dev/null +++ b/backend/tests/test_auth_rate_limit.py @@ -0,0 +1,48 @@ +"""登录 IP 限速（auth_api._login_rate_limited）：窗口内超限拦截、过期滑出、独立 IP 隔离。""" +from __future__ import annotations + +import pytest + +import app.auth_api as auth_api +from app.auth_api import _LOGIN_MAX_PER_WINDOW, _LOGIN_WINDOW, _login_attempts, _login_rate_limited + + +class _FakeClock: + """替换 auth_api 命名空间里的 time（不影响全局 time 模块）。""" + + def __init__(self): + self.now = 1000.0 + + def monotonic(self) -> float: + return self.now + + +@pytest.fixture() +def clock(monkeypatch): + c = _FakeClock() + monkeypatch.setattr(auth_api, "time", c) + _login_attempts.clear() + yield c + _login_attempts.clear() + + +def test_under_limit_passes(clock): + for _ in range(_LOGIN_MAX_PER_WINDOW): + assert _login_rate_limited("1.2.3.4") is False + assert _login_rate_limited("1.2.3.4") is True # 第 16 次被拦 + + +def test_window_slides(clock): + for _ in range(_LOGIN_MAX_PER_WINDOW): + _login_rate_limited("1.2.3.4") + assert _login_rate_limited("1.2.3.4") is True + # 窗口滑过：最早的尝试过期出窗，重新放行 + clock.now += _LOGIN_WINDOW + 0.1 + assert _login_rate_limited("1.2.3.4") is False + + +def test_ips_isolated(clock): + for _ in range(_LOGIN_MAX_PER_WINDOW): + _login_rate_limited("1.1.1.1") + assert _login_rate_limited("1.1.1.1") is True + assert _login_rate_limited("2.2.2.2") is False # 另一 IP 不受牵连 diff --git a/backend/tests/test_cache_local.py b/backend/tests/test_cache_local.py new file mode 100644 index 0000000..59a1619 --- /dev/null +++ b/backend/tests/test_cache_local.py @@ -0,0 +1,94 @@ +"""cache.py 本地层（不碰 Redis）：TTL 过期、容量淘汰、熔断冷却恢复。""" +from __future__ import annotations + +import pytest + +import app.cache as cache_mod +from app import cache + + +class _FakeClock: + """替换 cache 命名空间里的 time（不影响全局 time 模块）。""" + + def __init__(self): + self.now = 1000.0 + + def monotonic(self) -> float: + return self.now + + def time(self) -> float: + return self.now + + +@pytest.fixture(autouse=True) +def _reset_state(): + """每个用例干净的本地缓存状态。""" + cache._local_store.clear() + cache._local_bytes = 0 + cache._disabled_until = 0.0 + yield + cache._local_store.clear() + cache._local_bytes = 0 + cache._disabled_until = 0.0 + + +def test_local_set_get_roundtrip(): + cache.local_set("k", '{"a":1}', ttl=60) + assert cache.local_get("k") == '{"a":1}' + + +def test_local_get_miss(): + assert cache.local_get("nope") is None + + +def test_local_expiry(monkeypatch): + clock = _FakeClock() + monkeypatch.setattr(cache_mod, "time", clock) + cache.local_set("k", "v", ttl=10) + clock.now = 1005.0 + assert cache.local_get("k") == "v" + clock.now = 1101.0 # 过期 + assert cache.local_get("k") is None + assert "k" not in cache._local_store # 过期读取顺手清理 + + +def test_local_ttl_capped_at_120s(): + """本地层恒 ≤120s（多进程部署时最多比 Redis 多陈旧 120s 的约定）。""" + base = cache.time.monotonic() + cache.local_set("k", "v", ttl=99999) + ent = cache._local_store["k"] + assert ent[0] - base <= 120.0 + 5 # 相对当前 monotonic 的上限（留误差余量） + + +def test_local_entries_cap_evicts_oldest(): + for i in range(cache._LOCAL_MAX_ENTRIES + 5): + cache.local_set(f"k{i}", "v", ttl=60) + assert len(cache._local_store) <= cache._LOCAL_MAX_ENTRIES + # 先插入的（最旧）被近似 LRU 淘汰 + assert cache.local_get("k0") is None + assert cache.local_get(f"k{cache._LOCAL_MAX_ENTRIES + 4}") == "v" + + +def test_local_overwrite_releases_bytes(): + cache.local_set("k", "x" * 1000, ttl=60) + before = cache._local_bytes + cache.local_set("k", "y", ttl=60) + assert cache._local_bytes < before + assert cache.local_get("k") == "y" + + +def test_bail_cooldown_recovers(monkeypatch): + """熔断 60s：期间 _client 为 None，到期自动放行（Redis 未配置时也返回 None，但不熔断）。""" + clock = _FakeClock() + monkeypatch.setattr(cache_mod, "time", clock) + cache._bail() + assert cache._disabled_until > clock.now + clock.now += 59.0 + assert clock.monotonic() < cache._disabled_until # 仍在熔断期 + clock.now += 2.0 # 越过 60s 冷却 + assert clock.monotonic() >= cache._disabled_until # 恢复探测资格 + + +def test_digest_stable_and_distinct(): + assert cache.digest("a", 1, None) == cache.digest("a", 1, None) + assert cache.digest("a", 1) != cache.digest("a", 2) diff --git a/backend/tests/test_events.py b/backend/tests/test_events.py new file mode 100644 index 0000000..b4c67fd --- /dev/null +++ b/backend/tests/test_events.py @@ -0,0 +1,60 @@ +"""事件回测纯函数：入场/出场索引语义、汇总统计（含空样本与分年）。""" +from __future__ import annotations + +from datetime import datetime + +import pytest + +from app.backtest.events import _entry_exit_indices, _stats_block +from app.schemas import EventBacktestSpec + + +def _spec(holding_days: int = 5) -> EventBacktestSpec: + return EventBacktestSpec.model_validate({ + "holding_days": holding_days, + "entry": {"indicator": []}, + }) + + +def test_entry_exit_next_day_and_hold(): + spec = _spec(holding_days=5) + assert _entry_exit_indices(10, spec, n=100) == (11, 16) # 次日入，持有 5 日出 + + +def test_entry_exit_out_of_range_none(): + spec = _spec(holding_days=5) + # 出场索引越界（exit_i >= n） + assert _entry_exit_indices(94, spec, n=100) is None + assert _entry_exit_indices(99, spec, n=100) is None + + +def test_entry_exit_boundary_exact_fit(): + spec = _spec(holding_days=5) + # exit_i == n-1 恰好可用 + assert _entry_exit_indices(93, spec, n=100) == (94, 99) + + +def test_stats_block_empty(): + s = _stats_block([]) + assert s["samples"] == 0 and s["stocks"] == 0 + assert s["by_year"] == [] + assert s["win_rate"] == 0.0 + + +def test_stats_block_aggregates(): + trades = [ + {"ts_code": "000001.SZ", "ret_pct": 10.0, + "entry_date": datetime(2024, 1, 5), "entry_price": 10, "exit_price": 11}, + {"ts_code": "000001.SZ", "ret_pct": -4.0, + "entry_date": datetime(2024, 3, 6), "entry_price": 10, "exit_price": 9.6}, + {"ts_code": "600519.SH", "ret_pct": 2.0, + "entry_date": datetime(2023, 5, 10), "entry_price": 10, "exit_price": 10.2}, + ] + s = _stats_block(trades) + assert s["samples"] == 3 + assert s["stocks"] == 2 + assert s["mean_pct"] == pytest.approx((10.0 - 4.0 + 2.0) / 3, abs=1e-3) # 统计块保留 3 位小数 + assert s["win_rate"] == round(2 / 3 * 100, 2) + assert [y["year"] for y in s["by_year"]] == [2023, 2024] # 分年升序 + assert s["by_year"][1]["samples"] == 2 + assert s["max_pct"] == 10.0 and s["min_pct"] == -4.0 diff --git a/backend/tests/test_sync_utils.py b/backend/tests/test_sync_utils.py new file mode 100644 index 0000000..d643ed5 --- /dev/null +++ b/backend/tests/test_sync_utils.py @@ -0,0 +1,85 @@ +"""sync_utils 纯函数：取值清洗、日期格式化、限频重试语义。""" +from __future__ import annotations + +import math +from datetime import timedelta + +import pytest + +from app.data import sync_utils + + +def test_s_clean(): + assert sync_utils.s_clean(" 平安银行 ") == "平安银行" + assert sync_utils.s_clean("") is None + assert sync_utils.s_clean(" ") is None + assert sync_utils.s_clean(None) is None + assert sync_utils.s_clean(float("nan")) is None # pandas NaN + + +def test_f_clean(): + assert sync_utils.f_clean("3.14") == 3.14 + assert sync_utils.f_clean(2) == 2.0 + assert sync_utils.f_clean(float("nan")) is None + assert sync_utils.f_clean(None) is None + assert sync_utils.f_clean("abc") is None + assert sync_utils.f_clean(math.inf) == math.inf # inf 非 NaN，原样保留 + + +def test_d8_iso(): + assert sync_utils.d8_iso("20240102") == "2024-01-02" + assert sync_utils.d8_iso(20240102) == "2024-01-02" + assert sync_utils.d8_iso(None) is None + assert sync_utils.d8_iso("") is None + + +def test_fresh(): + now = sync_utils.utcnow() + assert sync_utils.fresh(now, days=7) is True + assert sync_utils.fresh(now - timedelta(days=8), days=7) is False + assert sync_utils.fresh(None, days=7) is False + + +def test_call_retry_passes_through_args(): + calls = [] + + def fn(a, b=0): + calls.append((a, b)) + return a + b + + assert sync_utils.call_retry(fn, 1, b=2) == 3 + assert calls == [(1, 2)] + + +def test_call_retry_rate_limit_retries_once(monkeypatch): + """「每分钟」级频率超限等 62s 重试一次；重试成功则返回结果。""" + monkeypatch.setattr(sync_utils.time, "sleep", lambda s: None) + calls = [] + + def fn(): + calls.append(1) + if len(calls) == 1: + raise RuntimeError("抱歉，您每分钟最多访问该接口5次") + return "ok" + + assert sync_utils.call_retry(fn) == "ok" + assert len(calls) == 2 + + +def test_call_retry_hourly_limit_raises(monkeypatch): + """小时级限频不重试，直接抛出。""" + monkeypatch.setattr(sync_utils.time, "sleep", lambda s: None) + + def fn(): + raise RuntimeError("您每小时最多访问该接口10次") + + with pytest.raises(RuntimeError): + sync_utils.call_retry(fn) + + +def test_call_retry_other_errors_raise(): + def fn(): + raise ValueError("数据源故障") + + with pytest.raises(ValueError): + sync_utils.call_retry(fn) diff --git a/backend/tests/test_trades_parser.py b/backend/tests/test_trades_parser.py new file mode 100644 index 0000000..b75e7e0 --- /dev/null +++ b/backend/tests/test_trades_parser.py @@ -0,0 +1,101 @@ +"""交割单解析器：四类真实导出格式 + 边界（转账/配号/利息跳过、费用合计去重、日期多格式）。 + +移植自 scripts/test_trades_parser.py（已删），不碰数据库，直接调 app.trades.parse_statement。 +""" +from __future__ import annotations + +import io +from datetime import datetime as dt + +import pytest +from fastapi import HTTPException +from openpyxl import Workbook + +from app.trades import parse_statement + + +def test_tdx_gbk_tabs(): + """通达信式：GBK + 制表符 + 标题行在前；跳过银行转存/利息归本。""" + tdx = ( + "交割单\n" + "股东账号: A123456789 起始日期: 20240102 终止日期: 20240105 币种: 人民币\n" + "\t交割日期\t业务名称\t证券代码\t证券名称\t成交价格\t成交数量\t成交金额\t手续费\t印花税\t过户费\t发生金额\t资金余额\t合同号\n" + "\t20240102\t证券买入\t600519\t贵州茅台\t1680.00\t100\t168000.00\t5.00\t0.00\t1.68\t-168006.68\t200000.00\t1000001\n" + "\t20240102\t银行转存\t\t\t\t\t\t\t\t\t50000.00\t250000.00\t\n" + "\t20240103\t证券卖出\t600519\t贵州茅台\t1700.50\t100\t170050.00\t5.00\t170.05\t1.70\t169873.25\t419873.25\t1000002\n" + "\t20240105\t利息归本\t\t\t\t\t\t\t\t\t1.25\t419874.50\t\n" + ) + r = parse_statement(tdx.encode("gbk"), "交割单.txt") + assert len(r.trades) == 2 + assert r.skipped_other == 2 + t0, t1 = r.trades[0], r.trades[1] + assert (t0.trade_date.isoformat(), t0.ts_code) == ("2024-01-02", "600519.SH") + assert t0.direction == "buy" and abs(t0.fee - 6.68) < 1e-9 + assert t1.direction == "sell" and abs(t1.fee - 176.75) < 1e-9 + assert t0.amount == 168000.0 + + +def test_hengsheng_csv_fee_total(): + """恒生柜台式：UTF-8 CSV，费用合计列不与其他费用列重复累加。""" + hs = ( + "序号,交收日期,证券代码,证券名称,交易类别,成交价格,成交数量,证券余额,成交金额,资金发生数,资金余额,流水序号,业务标志,业务名称,发生金额,后资金额,货币类别,费用合计,净佣金,规费,印花税,过户费,合同号\n" + "1,2024-06-07,000858,五粮液,证券买入,132.50,200,200,26500.00,-26505.80,73494.20,1,0101,证券买入,-26505.80,73494.20,人民币,5.80,4.20,1.60,0.00,0.00,66778001\n" + "2,2024-06-07,,,\t,,,,5120.00,78614.20,2,2041,银行转存,5120.00,78614.20,人民币,0,0,0,0,0,\n" + "3,2024-06-10,000858,五粮液,证券卖出,135.00,200,0,27000.00,26975.30,105589.50,3,0102,证券卖出,26975.30,105589.50,人民币,24.70,4.20,1.60,18.90,0.00,66779001\n" + ) + r = parse_statement(hs.encode("utf-8"), "hsi.csv") + assert len(r.trades) == 2 + assert abs(r.trades[1].fee - 24.70) < 1e-9 + assert r.trades[0].ts_code == "000858.SZ" + assert r.trades[0].trade_date.isoformat() == "2024-06-07" + + +def test_html_pseudo_xls(): + """HTML 伪 .xls（同花顺导出常见真身）：千分位金额、斜杠日期、创业板后缀。""" + html = """ + + + + + +
客户姓名测试
成交日期业务名称证券代码证券名称成交价格成交数量成交金额手续费
2024/03/15证券买入300750宁德时代182.3030054,690.0016.41
2024/03/18证券卖出300750宁德时代185.0030055,500.005.55
""" + r = parse_statement(html.encode("gbk"), "jiaogedan.xls") + assert len(r.trades) == 2 + assert r.trades[0].amount == 54690.0 + assert r.trades[0].ts_code == "300750.SZ" + assert r.trades[1].trade_date.isoformat() == "2024-03-18" + + +def test_amount_sign_direction(): + """无业务名称列（招商式）：发生金额正负判方向。""" + zh = ( + "证券名称,成交日期,成交价格,成交数量,发生金额,资金余额,合同编号\n" + "贵州茅台,20240102,1680.00,100,-168005.00,200000.00,SZ1000001\n" + "贵州茅台,20240103,1700.50,100,170049.50,370049.50,SZ1000002\n" + ) + r = parse_statement(zh.encode("utf-8"), "zszs.csv") + assert len(r.trades) == 2 + assert (r.trades[0].direction, r.trades[1].direction) == ("buy", "sell") + + +def test_xlsx_openpyxl(): + """xlsx（openpyxl 内存构造）：datetime 日期、科创板后缀、佣金+过户费合计。""" + wb = Workbook() + ws = wb.active + ws.append(["对账单", None, None]) + ws.append(["成交日期", "业务名称", "证券代码", "证券名称", "成交均价", "成交股数", "成交金额", "佣金", "过户费"]) + ws.append([dt(2024, 2, 28, 14, 35, 0), "证券买入", "688981", "中芯国际", 52.80, 200, 10560.00, 2.50, 1.06]) + ws.append([dt(2024, 3, 1, 9, 31, 0), "证券卖出", "688981", "中芯国际", 54.10, 200, 10820.00, 2.50, 1.06]) + buf = io.BytesIO() + wb.save(buf) + r = parse_statement(buf.getvalue(), "sm.xlsx") + assert len(r.trades) == 2 + assert r.trades[0].trade_date.isoformat() == "2024-02-28" + assert r.trades[0].ts_code == "688981.SH" + assert abs(r.trades[0].fee - 3.56) < 1e-9 + + +def test_garbage_input_422(): + with pytest.raises(HTTPException) as ei: + parse_statement("随便一串不是交割单的文字,1,2,3".encode("utf-8"), "x.csv") + assert ei.value.status_code == 422 diff --git a/frontend/src/api/client.ts b/frontend/src/api/client.ts index fc030d2..5f8e2c8 100644 --- a/frontend/src/api/client.ts +++ b/frontend/src/api/client.ts @@ -288,6 +288,7 @@ export async function getStocks(params: { industry?: string; area?: string; watched_only?: boolean; + held_only?: boolean; sort?: string; order?: 'asc' | 'desc'; limit?: number; @@ -300,6 +301,7 @@ export async function getStocks(params: { if (params.industry) q.set('industry', params.industry); if (params.area) q.set('area', params.area); if (params.watched_only) q.set('watched_only', 'true'); + if (params.held_only) q.set('held_only', 'true'); if (params.sort) q.set('sort', params.sort); if (params.order) q.set('order', params.order); q.set('limit', String(params.limit ?? 100)); @@ -384,6 +386,24 @@ export async function removeWatchlist(tsCode: string): Promise { return (await res.json()) as string[]; } +export async function getHoldings(): Promise { + const res = await apiFetch('/api/holdings'); + if (!res.ok) throw new ApiError(await readError(res, `获取持仓股失败 (HTTP ${res.status})`), res.status); + return (await res.json()) as string[]; +} + +export async function addHolding(tsCode: string): Promise { + const res = await apiFetch('/api/holdings', { method: 'POST', body: JSON.stringify({ ts_code: tsCode }) }); + if (!res.ok) throw new ApiError(await readError(res, `加持仓失败 (HTTP ${res.status})`), res.status); + return (await res.json()) as string[]; +} + +export async function removeHolding(tsCode: string): Promise { + const res = await apiFetch(`/api/holdings/${encodeURIComponent(tsCode)}`, { method: 'DELETE' }); + if (!res.ok) throw new ApiError(await readError(res, `移除持仓失败 (HTTP ${res.status})`), res.status); + return (await res.json()) as string[]; +} + export async function getScreenerQueries(limit = 20): Promise { const res = await apiFetch(`/api/screener/queries?limit=${limit}`); if (!res.ok) throw new ApiError(await readError(res, `获取提问历史失败 (HTTP ${res.status})`), res.status); diff --git a/frontend/src/api/types.ts b/frontend/src/api/types.ts index c2c2f0f..1d3a9a3 100644 --- a/frontend/src/api/types.ts +++ b/frontend/src/api/types.ts @@ -260,6 +260,7 @@ export interface StockListItem { total_mv?: number | null; // 总市值（亿元） circ_mv?: number | null; // 流通市值（亿元） watched: boolean; + held: boolean; // 是否持仓（当前用户） } export interface StockListResponse { diff --git a/frontend/src/components/LimitBoard.vue b/frontend/src/components/LimitBoard.vue new file mode 100644 index 0000000..e782f27 --- /dev/null +++ b/frontend/src/components/LimitBoard.vue @@ -0,0 +1,253 @@ + + + diff --git a/frontend/src/components/StockDetailOverlay.vue b/frontend/src/components/StockDetailOverlay.vue index 349872c..6bf5a6e 100644 --- a/frontend/src/components/StockDetailOverlay.vue +++ b/frontend/src/components/StockDetailOverlay.vue @@ -1,8 +1,8 @@ + + diff --git a/frontend/src/views/StocksView.vue b/frontend/src/views/StocksView.vue index 3807763..791c57a 100644 --- a/frontend/src/views/StocksView.vue +++ b/frontend/src/views/StocksView.vue @@ -1,7 +1,7 @@