feat: support historical price limits and share capital

backend/app/api/financials.py

@@ -8,7 +8,7 @@ from fastapi import APIRouter, HTTPException, Request
 from fastapi.responses import StreamingResponse
 from pydantic import BaseModel
 
-from app.services.financial_sync import get_financial_df
+from app.services.financial_sync import FINANCIAL_TABLES, get_financial_df
 from app.services.financial_analyzer import analyze_financials_stream
 from app.services import ai_reports
 from app.tickflow.capabilities import Cap
@@ -43,7 +43,7 @@ def financial_status(request: Request):
     data_dir = request.app.state.repo.store.data_dir
     tables = {}
 
-    for table in ("metrics", "income", "balance_sheet", "cash_flow"):
+    for table in FINANCIAL_TABLES:
         path = data_dir / "financials" / table / "part.parquet"
         if path.exists():
             try:
@@ -126,18 +126,32 @@ def get_cash_flow(request: Request, symbol: str | None = None):
     return {"data": df.to_dicts()}
 
 
+@router.get("/shares")
+def get_shares(request: Request, symbol: str | None = None):
+    """查询历史股本表。"""
+    capset = request.app.state.capabilities
+    _require_financial(capset)
+
+    df = get_financial_df(request.app.state.repo.store.data_dir, "shares")
+    if df.is_empty():
+        return {"data": []}
+    if symbol:
+        df = df.filter(pl.col("symbol") == symbol)
+    return {"data": df.to_dicts()}
+
+
 @router.post("/sync/{table}")
 def sync_table(request: Request, table: str):
     """手动触发同步(立即返回,后台异步执行)。
 
-    table: metrics / income / balance_sheet / cash_flow / all
+    table: metrics / income / balance_sheet / cash_flow / shares / all
     同步在后台线程执行,全量同步需数分钟。本接口立即返回 started 状态,
     前端通过轮询 GET /status 的 syncing 字段观察进度。
     """
     capset = request.app.state.capabilities
     _require_financial(capset)
 
-    valid_tables = {"metrics", "income", "balance_sheet", "cash_flow", "all"}
+    valid_tables = {*FINANCIAL_TABLES, "all"}
     if table not in valid_tables:
         raise HTTPException(400, f"invalid table: {table}, expected one of {valid_tables}")
 
@@ -161,7 +175,7 @@ class AnalyzeRequest(BaseModel):
 async def analyze_financials(request: Request, req: AnalyzeRequest):
     """AI 财务分析 — SSE 流式返回。
 
-    后端读取该标的 4 张财务表 → 注入 CFA 分析师级提示词 → 流式调用 LLM →
+    后端读取该标的财务报表与股本表 → 注入 CFA 分析师级提示词 → 流式调用 LLM →
     逐 chunk 以 SSE 形式推给前端(JSON per line, 非 text/event-stream,
     以便前端用 ReadableStream 逐行解析,更简单可靠)。
     """

backend/app/api/kline.py

@@ -2,6 +2,7 @@
 from __future__ import annotations
 
 import logging
+import math
 from datetime import date, timedelta
 from typing import Optional
 
@@ -9,6 +10,7 @@ from fastapi import APIRouter, HTTPException, Query, Request
 
 from app.indicators.pipeline import compute_enriched, compute_enriched_single
 from app.market_time import cn_now, cn_today
+from app.price_limits import is_risk_warning_name, price_limit_pct
 from app.services import kline_sync
 
 logger = logging.getLogger(__name__)
@@ -133,6 +135,67 @@ def _get_asset_info(repo, symbol: str, asset_type: str) -> dict:
         return {}
 
 
+def _get_price_limit_info(
+    repo,
+    symbol: str,
+    trade_date: date,
+    asset_type: str,
+    instrument_name: str | None,
+) -> dict | None:
+    """Return the date-aware limit rule and today's authoritative prices."""
+    if asset_type == "index":
+        return None
+
+    info = {
+        "rate": price_limit_pct(
+            symbol,
+            trade_date,
+            is_risk_warning=(
+                asset_type == "stock" and is_risk_warning_name(instrument_name)
+            ),
+        ),
+        "limit_up": None,
+        "limit_down": None,
+        "source": "rule",
+    }
+    if trade_date != cn_today():
+        return info
+
+    try:
+        import polars as pl
+
+        instruments = repo.get_instruments_asset(asset_type)
+        available = [
+            column
+            for column in ("symbol", "limit_up", "limit_down")
+            if column in instruments.columns
+        ]
+        if "symbol" not in available or len(available) == 1:
+            return info
+        hit = instruments.filter(pl.col("symbol") == symbol).select(available).head(1)
+        row = hit.to_dicts()[0] if not hit.is_empty() else None
+    except Exception:
+        return info
+    if row is None:
+        return info
+
+    has_authoritative_price = False
+    for field in ("limit_up", "limit_down"):
+        value = row.get(field)
+        if value is None:
+            continue
+        try:
+            numeric = float(value)
+        except (TypeError, ValueError):
+            continue
+        if math.isfinite(numeric) and 0 < numeric < 10_000:
+            info[field] = numeric
+            has_authoritative_price = True
+    if has_authoritative_price:
+        info["source"] = "instrument"
+    return info
+
+
 @router.get("/daily")
 def get_daily(
     request: Request,
@@ -532,11 +595,18 @@ def get_minute(
         # 本地无任何分钟K，尝试从 TickFlow 拉取当天
         trade_date = cn_today()
         df = kline_sync.fetch_minute_single(symbol, trade_date)
+        price_limit = _get_price_limit_info(
+            repo, symbol, trade_date, asset_type, stock_name,
+        )
         return {
             "symbol": symbol, "name": stock_name, "stock_info": stock_info,
             "date": str(trade_date), "rows": df.to_dicts(), "source": "live",
+            "price_limit": price_limit,
         }
 
+    price_limit = _get_price_limit_info(
+        repo, symbol, trade_date, asset_type, stock_name,
+    )
     df = repo.get_minute(symbol, trade_date, asset_type=asset_type)
 
     # 完整交易日应有 240 条分钟K；如果是今天(盘中)，期望条数按已交易分钟估算
@@ -562,6 +632,7 @@ def get_minute(
         return {
             "symbol": symbol, "name": stock_name, "stock_info": stock_info,
             "date": str(trade_date), "rows": df.to_dicts(), "source": "local",
+            "price_limit": price_limit,
         }
 
     # 本地不完整或无数据 → 从 TickFlow 实时拉取
@@ -570,6 +641,7 @@ def get_minute(
         "symbol": symbol, "name": stock_name, "stock_info": stock_info,
         "date": str(trade_date), "rows": live_df.to_dicts(),
         "source": "live" if not live_df.is_empty() else "none",
+        "price_limit": price_limit,
     }
 
 
@@ -1013,4 +1085,3 @@ async def rebuild_enriched(request: Request):
 # 长时间任务专用线程池（隔离于 FastAPI 默认线程池，防止阻塞请求处理）
 import concurrent.futures as _cf
 _long_task_executor = _cf.ThreadPoolExecutor(max_workers=2, thread_name_prefix="long-task")
-

backend/app/backtest/engine.py

@@ -354,6 +354,11 @@ class BacktestEngine:
                 needed={"signal_limit_up", "signal_limit_down"}
                 if matrix_native
                 else set(feature_plan.signal_columns),
+                historical_shares=(
+                    self.repo.get_historical_shares()
+                    if asset_type == "stock" and self.repo is not None
+                    else None
+                ),
             )
             join_cols = ["symbol"] if "symbol" in instruments.columns else []
             join_cols.extend(

backend/app/backtest/matrix.py

@@ -11,7 +11,7 @@ import time
 import uuid
 import weakref
 from collections import OrderedDict
-from collections.abc import Callable, Iterator, Mapping
+from collections.abc import Callable, Iterator, Mapping, Sequence
 from contextlib import contextmanager, nullcontext
 from contextvars import ContextVar
 from dataclasses import dataclass, field
@@ -27,6 +27,12 @@ import pyarrow.compute as pc
 import pyarrow.dataset as pads
 
 from app.backtest.minute_trigger import build_minute_exit_reference
+from app.price_limits import (
+    MAIN_BOARD_ST_LIMIT_CHANGE_DATE,
+    numpy_limit_pct_vectors,
+    numpy_limit_price,
+    write_numpy_price_limit_matrix,
+)
 
 try:
     from numba import njit, prange
@@ -43,7 +49,7 @@ except ImportError:
     prange = range
 
 _MATRIX_CACHE_VERSION = 1
-_DIRECT_MATRIX_LOADER_VERSION = 3
+_DIRECT_MATRIX_LOADER_VERSION = 4
 _MATRIX_AXIS_INDEX_VERSION = 1
 _ARROW_BATCH_SIZE = 131_072
 _SCORE_ASSET_CHUNK_SIZE = 256
@@ -589,6 +595,8 @@ def build_market_data_matrix(
     wanted_fields = set(field_columns or ()) - core_columns
     fields: dict[str, np.ndarray] = {}
     for column in sorted(wanted_fields):
+        if column == "price_limit_pct":
+            continue
         if column in panel.columns and panel[column].dtype.is_numeric():
             fields[column] = float_matrix(column)
         elif column == "raw_close":
@@ -604,6 +612,16 @@ def build_market_data_matrix(
             if not names[int(aid)] and row_names[row]:
                 names[int(aid)] = str(row_names[row])
 
+    if "price_limit_pct" in wanted_fields:
+        trading_dates = unique_timestamps.cast(pl.Date).to_list()
+        fields["price_limit_pct"] = write_numpy_price_limit_matrix(
+            np.empty(shape, dtype=np.float32),
+            trading_dates,
+            symbol_values,
+            names,
+            valid=np.isfinite(close),
+        )
+
     timestamp_labels = tuple(str(value)[:19] for value in unique_timestamps.to_numpy())
     timestamps = _timestamp_int64(unique_timestamps)
     session_dates = unique_timestamps.cast(pl.Date).to_numpy()
@@ -857,7 +875,9 @@ def _resolve_matrix_storage_fields(
     parquet_fields = sorted(
         name
         for name in wanted_fields
-        if name in available and _arrow_numeric(dataset.schema.field(name).type)
+        if name != "price_limit_pct"
+        and name in available
+        and _arrow_numeric(dataset.schema.field(name).type)
     )
     instrument_columns = set(instruments.columns) if instruments is not None else set()
     matrix_fields = set(parquet_fields)
@@ -874,6 +894,8 @@ def _resolve_matrix_storage_fields(
         matrix_fields.add("turnover_rate")
         if "turnover_rate" not in parquet_fields and "float_shares" in instrument_columns:
             vector_fields.add("float_shares")
+    if "price_limit_pct" in wanted_fields:
+        matrix_fields.add("price_limit_pct")
     resolved = matrix_fields | vector_fields
     unresolved = wanted_fields - resolved
     if unresolved:
@@ -940,6 +962,14 @@ def _build_market_data_matrix_from_dataset(
         parquet_fields=parquet_fields,
         vector_fields=vector_fields,
     )
+    if "price_limit_pct" in fields:
+        write_numpy_price_limit_matrix(
+            fields["price_limit_pct"],
+            actual_dates,
+            actual_symbols,
+            names,
+            valid=seen,
+        )
     for name in vector_fields:
         fields[name] = np.where(seen, fields[name], np.nan).astype(np.float32, copy=False)
     tradable = _tradable_matrix(
@@ -954,6 +984,7 @@ def _build_market_data_matrix_from_dataset(
         arrays["close"],
         raw_close,
         seen,
+        actual_dates,
         actual_symbols,
         names,
         latest_limits,
@@ -1085,6 +1116,14 @@ def _build_market_data_matrix_cache_from_dataset(
             vector_fields=vector_fields,
         )
         _mask_unseen_staging_fields(fields, seen)
+        if "price_limit_pct" in fields:
+            write_numpy_price_limit_matrix(
+                fields["price_limit_pct"],
+                actual_dates,
+                actual_symbols,
+                names,
+                valid=seen,
+            )
         _write_tradable_matrix(
             arrays["tradable"],
             arrays["open"],
@@ -1097,6 +1136,7 @@ def _build_market_data_matrix_cache_from_dataset(
             arrays["close"],
             fields.get("raw_close", arrays["close"]),
             seen,
+            actual_dates,
             actual_symbols,
             names,
             latest_limits,
@@ -2060,6 +2100,7 @@ def _limit_lock_matrices(
     close: np.ndarray,
     raw_close: np.ndarray,
     seen: np.ndarray,
+    trading_dates: Sequence[date],
     symbols: list[str],
     names: list[str],
     latest_limits: Mapping[str, np.ndarray],
@@ -2073,21 +2114,21 @@ def _limit_lock_matrices(
     down_locked = out_down if out_down is not None else np.zeros(shape, dtype=np.uint8)
     if up_locked.shape != shape or down_locked.shape != shape:
         raise ValueError("limit lock output shape mismatch")
+    if len(trading_dates) != shape[0]:
+        raise ValueError("price-limit date axis mismatch")
     up_locked.fill(0)
     down_locked.fill(0)
-    board_pct = np.full(shape[1], 0.10, dtype=np.float64)
-    for asset_id, symbol in enumerate(symbols):
-        if symbol.startswith(("300", "301", "688", "689")):
-            board_pct[asset_id] = 0.20
-        elif symbol.endswith(".BJ"):
-            board_pct[asset_id] = 0.30
-        elif "ST" in names[asset_id]:
-            board_pct[asset_id] = 0.05
+    legacy_pct, current_pct = numpy_limit_pct_vectors(symbols, names)
 
     previous_close = np.full(shape[1], np.nan, dtype=np.float64)
     previous_raw = np.full(shape[1], np.nan, dtype=np.float64)
     previous_adjustment = np.full(shape[1], np.nan, dtype=np.float64)
     for time_id in range(shape[0]):
+        limit_pct = (
+            legacy_pct
+            if trading_dates[time_id] < MAIN_BOARD_ST_LIMIT_CHANGE_DATE
+            else current_pct
+        )
         present = seen[time_id]
         current_close = close[time_id].astype(np.float64, copy=False)
         current_raw = raw_close[time_id].astype(np.float64, copy=False)
@@ -2112,8 +2153,8 @@ def _limit_lock_matrices(
             & (current_raw > 0)
         )
         if valid.any():
-            up_price = _numpy_limit_price(reference, board_pct, up=True)
-            down_price = _numpy_limit_price(reference, board_pct, up=False)
+            up_price = numpy_limit_price(reference, limit_pct, up=True)
+            down_price = numpy_limit_price(reference, limit_pct, up=False)
             if apply_latest_limits and time_id == shape[0] - 1:
                 latest_up = latest_limits["limit_up"]
                 latest_down = latest_limits["limit_down"]
@@ -2134,23 +2175,6 @@ def _limit_lock_matrices(
     return up_locked, down_locked
 
 
-def _numpy_limit_price(
-    previous: np.ndarray,
-    limit_pct: np.ndarray,
-    *,
-    up: bool,
-) -> np.ndarray:
-    sign = 1 if up else -1
-    numerator = np.rint((1.0 + sign * limit_pct) * 100.0).astype(np.int64)
-    result = np.full(previous.shape, np.nan, dtype=np.float64)
-    finite = np.isfinite(previous)
-    cents = np.floor(previous[finite] * 100.0 + 0.5).astype(np.int64)
-    result[finite] = (
-        ((cents * numerator[finite] + 50) // 100).astype(np.float64) / 100.0
-    )
-    return result
-
-
 def make_signal_matrix(
     shape: tuple[int, int],
     *,

backend/app/data_providers/custom/provider.py

@@ -130,11 +130,11 @@ class GenericHTTPProvider:
         self,
         table: str,
         symbols: list[str],
-        latest_only: bool = True,  # noqa: ARG002
+        latest_only: bool = True,
     ) -> pl.DataFrame:
-        """拉取财务数据。table ∈ {metrics, income, balance_sheet, cash_flow}。
+        """拉取财务数据。table 包含四张财务报表及 shares 股本表。
 
-        custom 源用一个 'financial' dataset 配置覆盖 4 张表; 请求时把 table 作为参数传给上游,
+        custom 源用一个 'financial' dataset 配置覆盖全部财务表; 请求时把 table 作为参数传给上游,
         上游根据 table 返回对应数据。字段由数据源决定, 这里只确保有 symbol 列。
         """
         cfg = self._dataset("financial")
@@ -142,9 +142,12 @@ class GenericHTTPProvider:
         chunks = chunked(symbols, cfg.batch)
         for i, chunk in enumerate(chunks):
             sleep_between_batches(i, cfg.rpm)
-            # 把 table 注入到请求参数 (上游据此区分 4 张表)
+            # 把 table 注入到请求参数 (上游据此区分财务表)
             extra_params = {**cfg.params, "table": table}
             extra_body = {**cfg.body, "table": table}
+            if table == "shares":
+                extra_params["latest"] = latest_only
+                extra_body["latest"] = latest_only
             rows = self._request_rows(
                 cfg, symbols=chunk,
                 override_params=extra_params, override_body=extra_body,

backend/app/indicators/pipeline.py

@@ -22,7 +22,14 @@ from pathlib import Path
 import polars as pl
 
 from app.config import settings
+from app.market_time import cn_today
 from app.parquet import scan_daily_parquet, scan_enriched_parquet, scan_parquet_compat
+from app.price_limits import (
+    polars_is_risk_warning_name,
+    polars_limit_price,
+    polars_price_limit_pct,
+)
+from app.share_capital import apply_historical_float_shares, load_share_history
 
 logger = logging.getLogger(__name__)
 
@@ -197,23 +204,6 @@ def _math_half_up(expr: pl.Expr, decimals: int = 2) -> pl.Expr:
     return (expr * factor + 0.5).floor() / factor
 
 
-def _limit_price(prev: pl.Expr, limit_pct: pl.Expr, up: bool) -> pl.Expr:
-    """用「分」为单位的整数算术计算涨跌停价，规避浮点精度问题。
-
-    交易所涨跌停价 = round(prev × (1 ± limit), 2)，标准四舍五入。
-    若直接用浮点 prev × (1 ± limit) 会丢精度：
-      18.90 × 0.95 = 17.955，浮点存储为 17.954999..., 四舍五入后得 17.95（错）。
-    本函数先把 prev 转成整数「分」(round 到分避免输入含厘误差)，
-    再用整数系数 105/95、110/90、120/80、130/70 相乘后四舍五入回元，全程不丢精度。
-    """
-    sign = 1 if up else -1
-    # limit_pct ∈ {0.05, 0.10, 0.20, 0.30} → 系数分子 105/95、110/90、120/80、130/70
-    num = ((1 + sign * limit_pct) * 100).cast(pl.Int64)  # 105, 110, 120, 130 等
-    cents = (prev * 100 + 0.5).floor().cast(pl.Int64)     # 价格转「分」(四舍五入到分)
-    # cents × num / 100, 四舍五入到分(加 50)
-    return (((cents * num + 50) // 100) / 100)
-
-
 def _apply_adj_factor(raw: pl.DataFrame, factors: pl.DataFrame) -> pl.DataFrame:
     """对 raw K 线应用前复权 (forward adjustment)。
 
@@ -647,6 +637,7 @@ def compute_limit_signals(
     df: pl.DataFrame,
     instruments: pl.DataFrame,
     needed: set[str] | None = None,
+    historical_shares: pl.DataFrame | None = None,
 ) -> pl.DataFrame:
     """计算涨跌停相关信号。
 
@@ -691,13 +682,19 @@ def compute_limit_signals(
     if need_price_limits and "name" in instruments.columns:
         st_flag = (
             instruments
-            .select("symbol", pl.col("name").str.contains("ST").alias("_is_st"))
+            .select(
+                "symbol",
+                polars_is_risk_warning_name(pl.col("name")).alias("_is_st"),
+            )
             .unique(subset=["symbol"])
         )
         inst_subset = inst_subset.join(st_flag, on="symbol", how="left")
 
     df = df.join(inst_subset, on="symbol", how="left", suffix="_inst")
 
+    if "turnover_rate" in want:
+        df = apply_historical_float_shares(df, historical_shares, today=cn_today())
+
     # 计算换手率(%) = volume(手) * 10000 / float_shares(股)
     if "turnover_rate" in want and "float_shares" in df.columns and "volume" in df.columns:
         df = df.with_columns(
@@ -726,40 +723,21 @@ def compute_limit_signals(
         .alias("_prev_raw_close")
     )
 
-    # 板块涨跌停比例
-    is_chinext = pl.col("symbol").str.starts_with("300") | pl.col("symbol").str.starts_with("301")
-    is_star = pl.col("symbol").str.starts_with("688") | pl.col("symbol").str.starts_with("689")
-    is_bj = pl.col("symbol").str.ends_with(".BJ")
-
+    is_risk_warning = pl.col("_is_st") if "_is_st" in df.columns else pl.lit(False)
     df = df.with_columns(
-        pl.when(is_chinext).then(0.20)
-        .when(is_star).then(0.20)
-        .when(is_bj).then(0.30)
-        .otherwise(0.10)
-        .alias("_board_pct")
+        polars_price_limit_pct(pl.col("symbol"), pl.col("date"), is_risk_warning)
+        .alias("_limit_pct")
     )
 
-    # ST → 5%, 但仅限主板风险警示股; 创业板/科创板/北交所 ST 保留各自板块限幅
-    # (注册制改革后 创业板 300/301、科创板 688/689 的 ST 仍执行 20%, 北交所 30%)。
-    if "_is_st" in df.columns:
-        df = df.with_columns(
-            pl.when(pl.col("_is_st").fill_null(False) & ~(is_chinext | is_star | is_bj))
-            .then(0.05)
-            .otherwise(pl.col("_board_pct"))
-            .alias("_limit_pct")
-        )
-    else:
-        df = df.with_columns(pl.col("_board_pct").alias("_limit_pct"))
-
     # 理论涨停价 = prev_close × (1 + limit_pct)  整数算术，避免浮点误差
     df = df.with_columns(
-        _limit_price(pl.col("_prev_raw_close"), pl.col("_limit_pct"), up=True)
+        polars_limit_price(pl.col("_prev_raw_close"), pl.col("_limit_pct"), up=True)
         .alias("_theoretical_limit_up")
     )
 
     # 理论跌停价 = prev_close × (1 - limit_pct)
     df = df.with_columns(
-        _limit_price(pl.col("_prev_raw_close"), pl.col("_limit_pct"), up=False)
+        polars_limit_price(pl.col("_prev_raw_close"), pl.col("_limit_pct"), up=False)
         .alias("_theoretical_limit_down")
     )
 
@@ -889,7 +867,7 @@ def compute_limit_signals(
         )
 
     # 清理临时列 + JOIN 引入的 instruments 列 (不存入 enriched)
-    cleanup = ["_prev_raw_close", "_board_pct", "_limit_pct",
+    cleanup = ["_prev_raw_close", "_limit_pct",
                "_theoretical_limit_up", "_theoretical_limit_down",
                "_effective_limit_up", "_effective_limit_down",
                "_grp_up", "_grp_down"]
@@ -910,7 +888,11 @@ def compute_limit_signals(
     return df
 
 
-def compute_all(df: pl.DataFrame, instruments: pl.DataFrame | None = None) -> pl.DataFrame:
+def compute_all(
+    df: pl.DataFrame,
+    instruments: pl.DataFrame | None = None,
+    historical_shares: pl.DataFrame | None = None,
+) -> pl.DataFrame:
     """从 OHLCV 计算全套指标 + 信号。一站式调用。
 
     输入: symbol, date, open, high, low, close, volume, amount, raw_close
@@ -918,7 +900,7 @@ def compute_all(df: pl.DataFrame, instruments: pl.DataFrame | None = None) -> pl
     df = compute_indicators(df)
     df = compute_signals(df)
     if instruments is not None and not instruments.is_empty():
-        df = compute_limit_signals(df, instruments)
+        df = compute_limit_signals(df, instruments, historical_shares=historical_shares)
 
     # 清理 NaN / Inf
     float_cols = [c for c in df.columns if df[c].dtype.is_float()]
@@ -954,6 +936,7 @@ def compute_enriched(
     raw: pl.DataFrame,
     factors: pl.DataFrame | None = None,
     instruments: pl.DataFrame | None = None,
+    historical_shares: pl.DataFrame | None = None,
 ) -> pl.DataFrame:
     """对原始日 K 应用前复权 + 全量计算指标 + 信号, 产出完整 enriched (含全部指标列)。
 
@@ -985,7 +968,11 @@ def compute_enriched(
     df = raw.sort(["symbol", "date"])
 
     # 全量计算指标 + 信号
-    df = compute_all(df, instruments=instruments)
+    df = compute_all(
+        df,
+        instruments=instruments,
+        historical_shares=historical_shares,
+    )
 
     return df
 
@@ -1041,6 +1028,7 @@ def run_pipeline(data_dir: Path | None = None,
         instruments = scan_parquet_compat(inst_glob, cast_options=_cast).collect()
     except Exception as e:  # noqa: BLE001
         logger.warning("instruments 读取失败: %s", e)
+    historical_shares = load_share_history(d)
 
     if new_dates_only:
         # ── 向后增量模式 ──
@@ -1083,7 +1071,12 @@ def run_pipeline(data_dir: Path | None = None,
             else:
                 raw_full = raw_new
 
-            enriched_new = compute_enriched(raw_full, factors=factors, instruments=instruments)
+            enriched_new = compute_enriched(
+                raw_full,
+                factors=factors,
+                instruments=instruments,
+                historical_shares=historical_shares,
+            )
 
             # 只保留新日期的行
             new_date_set = set()
@@ -1119,7 +1112,13 @@ def run_pipeline(data_dir: Path | None = None,
             if not raw_sym.is_empty():
                 factors_sym = factors.filter(pl.col("symbol").is_in(list(sym_set))) if not factors.is_empty() else factors
                 inst_sym = instruments.filter(pl.col("symbol").is_in(list(sym_set))) if not instruments.is_empty() else instruments
-                enriched_sym = compute_enriched(raw_sym, factors=factors_sym, instruments=inst_sym)
+                shares_sym = historical_shares.filter(pl.col("symbol").is_in(list(sym_set))) if not historical_shares.is_empty() else historical_shares
+                enriched_sym = compute_enriched(
+                    raw_sym,
+                    factors=factors_sym,
+                    instruments=inst_sym,
+                    historical_shares=shares_sym,
+                )
                 for date_df in enriched_sym.partition_by("date"):
                     dt = date_df["date"][0]
                     ds = dt.isoformat() if hasattr(dt, "isoformat") else str(dt)
@@ -1205,9 +1204,18 @@ def run_pipeline(data_dir: Path | None = None,
             inst_use.filter(pl.col("symbol").is_in(batch_syms))
             if not inst_use.is_empty() else inst_use
         )
+        batch_shares = (
+            historical_shares.filter(pl.col("symbol").is_in(batch_syms))
+            if not historical_shares.is_empty() else historical_shares
+        )
 
         # 计算
-        enriched = compute_enriched(raw, factors=batch_factors, instruments=batch_inst)
+        enriched = compute_enriched(
+            raw,
+            factors=batch_factors,
+            instruments=batch_inst,
+            historical_shares=batch_shares,
+        )
 
         if not enriched.is_empty():
             if symbols:
@@ -1233,7 +1241,7 @@ def run_pipeline(data_dir: Path | None = None,
                     date_buffers[ds].append(_select_storage_cols(date_df).sort(["symbol"]))
                     written += date_df.height
 
-        del raw, enriched, batch_factors, batch_inst
+        del raw, enriched, batch_factors, batch_inst, batch_shares
         gc.collect()
 
         logger.info("symbol 批次 %d/%d (%s ~ %s), 已处理 %d 行",
@@ -1654,7 +1662,10 @@ def _compute_limit_signals_today(df: pl.DataFrame, instruments: pl.DataFrame) ->
     if "name" in instruments.columns:
         st_flag = (
             instruments
-            .select("symbol", pl.col("name").str.contains("ST").alias("_is_st"))
+            .select(
+                "symbol",
+                polars_is_risk_warning_name(pl.col("name")).alias("_is_st"),
+            )
             .unique(subset=["symbol"])
         )
         inst_subset = inst_subset.join(st_flag, on="symbol", how="left")
@@ -1682,24 +1693,16 @@ def _compute_limit_signals_today(df: pl.DataFrame, instruments: pl.DataFrame) ->
         prev_raw = pl.col("close_right")
     else:
         prev_raw = pl.col("raw_close")
-    is_chinext = pl.col("symbol").str.starts_with("300") | pl.col("symbol").str.starts_with("301")
-    is_star = pl.col("symbol").str.starts_with("688") | pl.col("symbol").str.starts_with("689")
-    is_bj = pl.col("symbol").str.ends_with(".BJ")
-    limit_pct = (
-        pl.when(is_chinext).then(0.20)
-        .when(is_star).then(0.20)
-        .when(is_bj).then(0.30)
-        .otherwise(0.10)
-    )
-    if "_is_st" in df.columns:
-        # ST 5% 仅主板生效; 创业板/科创板/北交所 ST 保留板块限幅 (同 compute_limit_signals)
-        limit_pct = pl.when(
-            pl.col("_is_st").fill_null(False) & ~(is_chinext | is_star | is_bj)
-        ).then(0.05).otherwise(limit_pct)
-    limit_pct = limit_pct.alias("_limit_pct")
+    is_risk_warning = pl.col("_is_st") if "_is_st" in df.columns else pl.lit(False)
+    trade_date = pl.col("date") if "date" in df.columns else pl.lit(cn_today())
+    limit_pct = polars_price_limit_pct(
+        pl.col("symbol"),
+        trade_date,
+        is_risk_warning,
+    ).alias("_limit_pct")
 
-    limit_up_price = _limit_price(prev_raw, limit_pct, up=True)
-    limit_down_price = _limit_price(prev_raw, limit_pct, up=False)
+    limit_up_price = polars_limit_price(prev_raw, limit_pct, up=True)
+    limit_down_price = polars_limit_price(prev_raw, limit_pct, up=False)
 
     # 生效涨跌停价: 优先用维表权威值 (instruments.limit_up/down, 交易所级别精确价),
     # 维表缺失 (新股上市前 5 日: limit_up 为 null 或哨兵 100000) 回退自算理论价。

backend/app/price_limits.py

@@ -0,0 +1,158 @@
+"""A-share price-limit rules shared by indicators, backtests, and APIs."""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+from datetime import date
+
+import numpy as np
+import polars as pl
+
+MAIN_BOARD_ST_LIMIT_CHANGE_DATE = date(2026, 7, 6)
+
+MAIN_BOARD_LIMIT = 0.10
+LEGACY_MAIN_BOARD_ST_LIMIT = 0.05
+GROWTH_BOARD_LIMIT = 0.20
+BEIJING_BOARD_LIMIT = 0.30
+
+
+def is_risk_warning_name(name: str | None) -> bool:
+    return "ST" in str(name or "").upper()
+
+
+def board_limit_pct(symbol: str) -> float:
+    if symbol.endswith(".BJ"):
+        return BEIJING_BOARD_LIMIT
+    if symbol.startswith(("300", "301", "688", "689")):
+        return GROWTH_BOARD_LIMIT
+    return MAIN_BOARD_LIMIT
+
+
+def price_limit_pct(
+    symbol: str,
+    trade_date: date,
+    *,
+    is_risk_warning: bool = False,
+) -> float:
+    base = board_limit_pct(symbol)
+    if (
+        base == MAIN_BOARD_LIMIT
+        and is_risk_warning
+        and trade_date < MAIN_BOARD_ST_LIMIT_CHANGE_DATE
+    ):
+        return LEGACY_MAIN_BOARD_ST_LIMIT
+    return base
+
+
+def polars_price_limit_pct(
+    symbol: pl.Expr,
+    trade_date: pl.Expr,
+    is_risk_warning: pl.Expr,
+) -> pl.Expr:
+    """Return a vectorized Polars expression for the effective daily limit."""
+    is_growth = symbol.str.starts_with("300") | symbol.str.starts_with("301")
+    is_star = symbol.str.starts_with("688") | symbol.str.starts_with("689")
+    is_beijing = symbol.str.ends_with(".BJ")
+    is_non_main = is_growth | is_star | is_beijing
+    base = (
+        pl.when(is_growth | is_star).then(GROWTH_BOARD_LIMIT)
+        .when(is_beijing).then(BEIJING_BOARD_LIMIT)
+        .otherwise(MAIN_BOARD_LIMIT)
+    )
+    legacy_main_st = (
+        is_risk_warning.fill_null(False)
+        & ~is_non_main
+        & (trade_date < pl.lit(MAIN_BOARD_ST_LIMIT_CHANGE_DATE))
+    )
+    return (
+        pl.when(legacy_main_st).then(LEGACY_MAIN_BOARD_ST_LIMIT)
+        .otherwise(base)
+        .cast(pl.Float64)
+    )
+
+
+def polars_is_risk_warning_name(name: pl.Expr) -> pl.Expr:
+    """Return whether an instrument name contains the ST risk-warning marker."""
+    return name.fill_null("").str.to_uppercase().str.contains("ST", literal=True)
+
+
+def polars_limit_price(previous: pl.Expr, limit_pct: pl.Expr, *, up: bool) -> pl.Expr:
+    """Calculate exchange half-up prices with integer-cent arithmetic."""
+    sign = 1 if up else -1
+    numerator = ((1 + sign * limit_pct) * 100).round(0).cast(pl.Int64)
+    cents = (previous * 100 + 0.5).floor().cast(pl.Int64)
+    return ((cents * numerator + 50) // 100) / 100
+
+
+def numpy_limit_pct_vectors(
+    symbols: Sequence[str],
+    names: Sequence[str],
+) -> tuple[np.ndarray, np.ndarray]:
+    """Return pre/post-change vectors once; callers select one per date."""
+    current = np.fromiter(
+        (board_limit_pct(str(symbol)) for symbol in symbols),
+        dtype=np.float64,
+        count=len(symbols),
+    )
+    legacy = current.copy()
+    for asset_id, (_symbol, name) in enumerate(zip(symbols, names, strict=True)):
+        if current[asset_id] == MAIN_BOARD_LIMIT and is_risk_warning_name(name):
+            legacy[asset_id] = LEGACY_MAIN_BOARD_ST_LIMIT
+    return legacy, current
+
+
+def numpy_price_limit_matrix(
+    trading_dates: Sequence[date],
+    symbols: Sequence[str],
+    names: Sequence[str],
+) -> np.ndarray:
+    """Build a float32 time-by-asset matrix only for strategies that request it."""
+    result = np.empty((len(trading_dates), len(symbols)), dtype=np.float32)
+    return write_numpy_price_limit_matrix(result, trading_dates, symbols, names)
+
+
+def write_numpy_price_limit_matrix(
+    target: np.ndarray,
+    trading_dates: Sequence[date],
+    symbols: Sequence[str],
+    names: Sequence[str],
+    *,
+    valid: np.ndarray | None = None,
+) -> np.ndarray:
+    """Write date-aware limits directly into an existing matrix or memmap."""
+    expected_shape = (len(trading_dates), len(symbols))
+    if target.shape != expected_shape:
+        raise ValueError("price-limit output shape mismatch")
+    if valid is not None and valid.shape != expected_shape:
+        raise ValueError("price-limit validity mask shape mismatch")
+
+    legacy, current = numpy_limit_pct_vectors(symbols, names)
+    target[:] = current.astype(np.float32, copy=False)
+    legacy_rows = np.fromiter(
+        (value < MAIN_BOARD_ST_LIMIT_CHANGE_DATE for value in trading_dates),
+        dtype=bool,
+        count=len(trading_dates),
+    )
+    if legacy_rows.any():
+        target[legacy_rows] = legacy.astype(np.float32, copy=False)
+    if valid is not None:
+        target[~valid] = np.nan
+    return target
+
+
+def numpy_limit_price(
+    previous: np.ndarray,
+    limit_pct: np.ndarray,
+    *,
+    up: bool,
+) -> np.ndarray:
+    """NumPy counterpart of :func:`polars_limit_price`."""
+    sign = 1 if up else -1
+    numerator = np.rint((1.0 + sign * limit_pct) * 100.0).astype(np.int64)
+    result = np.full(previous.shape, np.nan, dtype=np.float64)
+    finite = np.isfinite(previous)
+    cents = np.floor(previous[finite] * 100.0 + 0.5).astype(np.int64)
+    result[finite] = (
+        ((cents * numerator[finite] + 50) // 100).astype(np.float64) / 100.0
+    )
+    return result

backend/app/services/financial_analyzer.py

@@ -1,6 +1,6 @@
 """AI 财务分析服务 — 读取个股财务数据 → 构建专业提示词 → 流式调用 LLM。
 
-职责: 拉取单只标的的 4 张财务表 → 转成紧凑 JSON → 拼装 CFA 分析师级系统提示词
+职责: 拉取单只标的的财务报表与股本表 → 转成紧凑 JSON → 拼装 CFA 分析师级系统提示词
        → 流式调用 OpenAI 兼容 API → 逐 chunk 吐给前端。
 
 不知道: HTTP、前端、配置持久化。
@@ -14,7 +14,7 @@ from typing import AsyncIterator
 
 import polars as pl
 
-from app.services.financial_sync import get_financial_df
+from app.services.financial_sync import FINANCIAL_TABLES, get_financial_df
 
 logger = logging.getLogger(__name__)
 
@@ -23,12 +23,12 @@ _MAX_PERIODS = 4
 
 
 def _load_stock_financials(data_dir: Path, symbol: str) -> dict[str, list[dict]]:
-    """读取该标的的 4 张财务表,返回 {table: [records...]}(按 period_end 降序,截取最新 N 期)。
+    """读取该标的财务数据,返回 {table: [records...]}(按 period_end 降序,截取最新 N 期)。
 
     数值统一做 NaN/Inf → null 清洗,保证 JSON 序列化不报错。
     """
     result: dict[str, list[dict]] = {}
-    for table in ("metrics", "income", "balance_sheet", "cash_flow"):
+    for table in FINANCIAL_TABLES:
         df = get_financial_df(data_dir, table)
         if df.is_empty():
             result[table] = []
@@ -60,7 +60,7 @@ def _load_stock_financials(data_dir: Path, symbol: str) -> dict[str, list[dict]]
 def _summarize(fins: dict[str, list[dict]]) -> str:
     """生成一行业务摘要,便于 LLM 快速把握数据全貌(行数/期数)。"""
     parts = []
-    for table in ("metrics", "income", "balance_sheet", "cash_flow"):
+    for table in FINANCIAL_TABLES:
         rows = fins.get(table, [])
         if rows:
             periods = [r.get("period_end") for r in rows if r.get("period_end")]

backend/app/services/financial_sync.py

@@ -21,8 +21,8 @@ logger = logging.getLogger(__name__)
 # 每个 API 请求最多 100 个标的
 _BATCH_SIZE = 100
 
-# 4 张财务表
-FINANCIAL_TABLES = ("metrics", "income", "balance_sheet", "cash_flow")
+# 财务报表 + 历史股本表
+FINANCIAL_TABLES = ("metrics", "income", "balance_sheet", "cash_flow", "shares")
 
 
 # ================================================================
@@ -52,35 +52,34 @@ def _financial_is_custom() -> bool:
     return custom_sources.provider_has_dataset(provider, "financial")
 
 
-def _sync_table(
+def _fetch_table(
     table: str,
     symbols: list[str],
-    data_dir: Path,
     capset: CapabilitySet,
     latest_only: bool = True,
-) -> int:
-    """同步单张财务表。返回写入的行数。"""
+) -> pl.DataFrame:
+    """通过当前财务数据源拉取一张标准化财务表。"""
     is_custom = _financial_is_custom()
     if not is_custom and not capset.has(Cap.FINANCIAL):
         logger.info("sync_%s skipped: no FINANCIAL capability", table)
-        return 0
+        return pl.DataFrame()
     if not symbols:
         logger.warning("sync_%s skipped: no symbols", table)
-        return 0
+        return pl.DataFrame()
 
     # 自定义数据源分流
     if is_custom:
         from app.services import preferences
         from app.data_providers import custom as custom_sources
-        provider = custom_sources.get_provider(preferences.get_financial_provider())
-        df = provider.get_financials(table, symbols, latest_only=latest_only)
+        try:
+            provider = custom_sources.get_provider(preferences.get_financial_provider())
+            df = provider.get_financials(table, symbols, latest_only=latest_only)
+        except Exception as e:  # noqa: BLE001
+            logger.warning("sync_%s custom provider failed: %s", table, e)
+            return pl.DataFrame()
         if df.is_empty() or "symbol" not in df.columns:
-            return 0
-        out_dir = data_dir / "financials" / table
-        out_dir.mkdir(parents=True, exist_ok=True)
-        df.write_parquet(out_dir / "part.parquet")
-        logger.info("sync_%s done via custom: %d records written", table, len(df))
-        return len(df)
+            return pl.DataFrame()
+        return df
 
     from app.tickflow.client import get_client
     tf = get_client()
@@ -91,7 +90,11 @@ def _sync_table(
         "income": tf.financials.income,
         "balance_sheet": tf.financials.balance_sheet,
         "cash_flow": tf.financials.cash_flow,
+        "shares": getattr(tf.financials, "shares", None),
     }[table]
+    if api_method is None:
+        logger.warning("sync_shares skipped: current TickFlow SDK does not support shares")
+        return pl.DataFrame()
 
     all_records: list[dict] = []
     total_batches = (len(symbols) + _BATCH_SIZE - 1) // _BATCH_SIZE
@@ -114,14 +117,16 @@ def _sync_table(
             logger.warning("sync_%s batch %d/%d failed: %s", table, batch_num, total_batches, e)
 
     if not all_records:
-        return 0
+        return pl.DataFrame()
 
     df = pl.DataFrame(all_records)
-    if df.is_empty():
-        return 0
+    if df.is_empty() or "symbol" not in df.columns:
+        return pl.DataFrame()
+    return df
 
-    # 确保 symbol 列存在
-    if "symbol" not in df.columns:
+
+def _write_table(table: str, df: pl.DataFrame, data_dir: Path) -> int:
+    if df.is_empty() or "symbol" not in df.columns:
         return 0
 
     # 写入 Parquet (全量覆盖)
@@ -134,6 +139,60 @@ def _sync_table(
     return len(df)
 
 
+def _sync_table(
+    table: str,
+    symbols: list[str],
+    data_dir: Path,
+    capset: CapabilitySet,
+    latest_only: bool = True,
+) -> int:
+    """同步单张财务表。返回写入的行数。"""
+    return _write_table(
+        table,
+        _fetch_table(table, symbols, capset, latest_only=latest_only),
+        data_dir,
+    )
+
+
+def _merge_share_history(*frames: pl.DataFrame) -> pl.DataFrame:
+    valid = [
+        frame
+        for frame in frames
+        if not frame.is_empty() and {"symbol", "period_end"} <= set(frame.columns)
+    ]
+    if not valid:
+        return pl.DataFrame()
+    return (
+        pl.concat(valid, how="diagonal_relaxed")
+        .filter(pl.col("symbol").is_not_null() & pl.col("period_end").is_not_null())
+        .unique(subset=["symbol", "period_end"], keep="last")
+        .sort(["symbol", "period_end"])
+    )
+
+
+def _sync_shares_for_symbols(
+    symbols: list[str],
+    data_dir: Path,
+    capset: CapabilitySet,
+) -> int:
+    """首次拉全量股本历史，后续更新最新记录并补齐新增标的历史。"""
+    existing = get_financial_df(data_dir, "shares")
+    if existing.is_empty() or not {"symbol", "period_end"} <= set(existing.columns):
+        return _sync_table("shares", symbols, data_dir, capset, latest_only=False)
+
+    existing_symbols = set(existing["symbol"].drop_nulls().to_list())
+    missing_symbols = [symbol for symbol in symbols if symbol not in existing_symbols]
+    missing_history = (
+        _fetch_table("shares", missing_symbols, capset, latest_only=False)
+        if missing_symbols
+        else pl.DataFrame()
+    )
+    current_symbols = [symbol for symbol in symbols if symbol in existing_symbols]
+    latest = _fetch_table("shares", current_symbols, capset, latest_only=True)
+    merged = _merge_share_history(existing, missing_history, latest)
+    return _write_table("shares", merged, data_dir)
+
+
 def sync_metrics(data_dir: Path, capset: CapabilitySet) -> int:
     """同步核心财务指标 (metrics)。"""
     symbols = _get_symbols(data_dir)
@@ -158,16 +217,26 @@ def sync_cash_flow(data_dir: Path, capset: CapabilitySet) -> int:
     return _sync_table("cash_flow", symbols, data_dir, capset, latest_only=True)
 
 
+def sync_shares(data_dir: Path, capset: CapabilitySet) -> int:
+    """同步历史股本表。"""
+    symbols = _get_symbols(data_dir)
+    return _sync_shares_for_symbols(symbols, data_dir, capset)
+
+
 def sync_all(data_dir: Path, capset: CapabilitySet) -> dict[str, int]:
     """同步所有财务表。返回 {table: rows}。"""
-    if not capset.has(Cap.FINANCIAL):
+    if not capset.has(Cap.FINANCIAL) and not _financial_is_custom():
         logger.info("sync_all financials skipped: no FINANCIAL capability")
         return {}
 
     symbols = _get_symbols(data_dir)
     results: dict[str, int] = {}
     for table in FINANCIAL_TABLES:
-        results[table] = _sync_table(table, symbols, data_dir, capset, latest_only=True)
+        results[table] = (
+            _sync_shares_for_symbols(symbols, data_dir, capset)
+            if table == "shares"
+            else _sync_table(table, symbols, data_dir, capset, latest_only=True)
+        )
 
     # 同步完成后注册 DuckDB 视图
     _refresh_financials_views(data_dir)
@@ -187,6 +256,7 @@ def _refresh_financials_views(data_dir: Path) -> None:
         "financials_income": f"{d}/financials/income/*.parquet",
         "financials_balance_sheet": f"{d}/financials/balance_sheet/*.parquet",
         "financials_cash_flow": f"{d}/financials/cash_flow/*.parquet",
+        "financials_shares": f"{d}/financials/shares/*.parquet",
     }
     for name, path in views.items():
         out = data_dir / "financials" / name.replace("financials_", "") / "part.parquet"
@@ -213,7 +283,7 @@ def get_financial_df(data_dir: Path, table: str) -> pl.DataFrame:
 # ================================================================
 
 class FinancialScheduler:
-    """独立调度器: 每周同步 metrics, 每季度同步三张报表。"""
+    """独立调度器: 每周同步 metrics, 财务表支持手动同步。"""
 
     def __init__(self) -> None:
         self._task: asyncio.Task | None = None
@@ -238,7 +308,7 @@ class FinancialScheduler:
         # 即便 app.state.capabilities 已更新, 调度器仍报 "no FINANCIAL capability"。
         self._data_dir = data_dir
         self._capset = capset
-        if not capset.has(Cap.FINANCIAL):
+        if not capset.has(Cap.FINANCIAL) and not _financial_is_custom():
             logger.info("FinancialScheduler skipped: no FINANCIAL capability")
             return
         # 从持久化恢复上次同步时间: 重启后前端仍能显示真实最后同步时间,而非"尚未同步"
@@ -338,7 +408,7 @@ class FinancialScheduler:
     def _run_body(self, table: str | None) -> dict[str, int]:
         """同步逻辑本体(不加锁,假设调用方已持有 _is_syncing)。
 
-        table=None 同步全部 4 张表;否则只同步指定表。
+        table=None 同步全部财务表;否则只同步指定表。
         每张表完成立即更新 last_sync,让前端轮询 /status 能看到进度递增。
         """
         if table:
@@ -347,6 +417,7 @@ class FinancialScheduler:
                 "income": sync_income,
                 "balance_sheet": sync_balance_sheet,
                 "cash_flow": sync_cash_flow,
+                "shares": sync_shares,
             }.get(table)
             if not fn:
                 return {}
@@ -357,7 +428,11 @@ class FinancialScheduler:
         symbols = _get_symbols(self._data_dir)
         result: dict[str, int] = {}
         for t in FINANCIAL_TABLES:
-            result[t] = _sync_table(t, symbols, self._data_dir, self._capset, latest_only=True)
+            result[t] = (
+                _sync_shares_for_symbols(symbols, self._data_dir, self._capset)
+                if t == "shares"
+                else _sync_table(t, symbols, self._data_dir, self._capset, latest_only=True)
+            )
             self._record_sync(t)
         _refresh_financials_views(self._data_dir)
         return result

backend/app/services/quote_service.py

@@ -1394,7 +1394,16 @@ class QuoteService:
                         pass
                 instruments = self._repo.get_instruments() if asset_type == "stock" else None
 
-                enriched_full = compute_enriched(full_df, factors=factors, instruments=instruments)
+                enriched_full = compute_enriched(
+                    full_df,
+                    factors=factors,
+                    instruments=instruments,
+                    historical_shares=(
+                        self._repo.get_historical_shares()
+                        if asset_type == "stock"
+                        else None
+                    ),
+                )
                 enriched_today = enriched_full.filter(pl.col("date") == today)
 
             if enriched_today.is_empty():

backend/app/services/screener.py

@@ -182,7 +182,11 @@ class ScreenerService:
         # 计算涨跌停信号 (需要 instruments; 涨停为股票专有, ETF 跳过)
         instruments = self.repo.get_instruments_asset(self.asset_type)
         if self.asset_type == "stock" and instruments is not None and not instruments.is_empty():
-            df_full = compute_limit_signals(df_full, instruments)
+            df_full = compute_limit_signals(
+                df_full,
+                instruments,
+                historical_shares=self.repo.get_historical_shares(),
+            )
 
         # 只保留目标日期
         df_result = df_full.filter(pl.col("date") == target_date)
@@ -264,7 +268,11 @@ class ScreenerService:
 
         instruments = self.repo.get_instruments_asset(self.asset_type)
         if self.asset_type == "stock" and instruments is not None and not instruments.is_empty():
-            df_full = compute_limit_signals(df_full, instruments)
+            df_full = compute_limit_signals(
+                df_full,
+                instruments,
+                historical_shares=self.repo.get_historical_shares(),
+            )
 
         if instruments is not None and not instruments.is_empty():
             inst_cols = [c for c in ["symbol", "name", "total_shares", "float_shares"] if c in instruments.columns]

backend/app/share_capital.py

@@ -0,0 +1,110 @@
+"""历史股本解析。
+
+财务股本按公告日可用，历史缺失时回退 instruments 最新流通股本。
+"""
+from __future__ import annotations
+
+from datetime import date
+from pathlib import Path
+
+import polars as pl
+
+
+def load_share_history(data_dir: Path) -> pl.DataFrame:
+    """读取本地财务股本表；未同步或损坏时返回空表。"""
+    path = data_dir / "financials" / "shares" / "part.parquet"
+    if not path.exists():
+        return pl.DataFrame()
+    try:
+        shares = pl.read_parquet(path)
+        if not {"symbol", "period_end", "float_shares"} <= set(shares.columns):
+            return pl.DataFrame()
+        return shares
+    except Exception:
+        return pl.DataFrame()
+
+
+def apply_historical_float_shares(
+    rows: pl.DataFrame,
+    shares: pl.DataFrame | None,
+    *,
+    today: date,
+) -> pl.DataFrame:
+    """为行情行解析有效流通股本。
+
+    当日保留 rows.float_shares；历史日期使用公告日不晚于交易日的最新股本，
+    找不到历史记录时继续使用 rows.float_shares。
+    """
+    required = {"symbol", "date", "float_shares"}
+    if (
+        rows.is_empty()
+        or not required <= set(rows.columns)
+        or shares is None
+        or shares.is_empty()
+        or not {"symbol", "period_end", "float_shares"} <= set(shares.columns)
+    ):
+        return rows
+
+    def as_date_expr(column: str) -> pl.Expr:
+        dtype = shares.schema[column]
+        if dtype == pl.Utf8:
+            return pl.col(column).str.to_date(strict=False)
+        return pl.col(column).cast(pl.Date, strict=False)
+
+    available_date = as_date_expr("period_end")
+    if "announce_date" in shares.columns:
+        available_date = as_date_expr("announce_date").fill_null(available_date)
+
+    history = (
+        shares
+        .select(
+            pl.col("symbol").cast(pl.Utf8),
+            available_date.alias("_share_available_date"),
+            pl.col("period_end").cast(pl.Utf8).alias("_share_period_end"),
+            pl.col("float_shares").cast(pl.Float64, strict=False).alias("_historical_float_shares"),
+        )
+        .filter(
+            pl.col("symbol").is_not_null()
+            & pl.col("_share_available_date").is_not_null()
+            & (pl.col("_historical_float_shares") > 0)
+        )
+        .sort(["symbol", "_share_available_date", "_share_period_end"])
+        .unique(subset=["symbol", "_share_available_date"], keep="last")
+        .sort(["symbol", "_share_available_date"])
+    )
+    if history.is_empty():
+        return rows
+
+    resolved = (
+        rows
+        .with_row_index("_share_row_order")
+        .with_columns(
+            pl.col("symbol").cast(pl.Utf8),
+            pl.col("date").cast(pl.Date, strict=False).alias("_share_trade_date"),
+        )
+        .sort(["symbol", "_share_trade_date"])
+        .join_asof(
+            history,
+            left_on="_share_trade_date",
+            right_on="_share_available_date",
+            by="symbol",
+            strategy="backward",
+            check_sortedness=False,
+        )
+        .with_columns(
+            pl.when(pl.col("_share_trade_date") == pl.lit(today))
+            .then(pl.col("float_shares"))
+            .otherwise(
+                pl.coalesce("_historical_float_shares", "float_shares")
+            )
+            .alias("float_shares")
+        )
+        .sort("_share_row_order")
+    )
+    return resolved.drop(
+        "_share_row_order",
+        "_share_trade_date",
+        "_share_available_date",
+        "_share_period_end",
+        "_historical_float_shares",
+    )

backend/app/strategy/builtin/near_limit_up.py

@@ -62,33 +62,22 @@ ALERTS = []
 
 class NearLimitUpMatrixStrategy:
     def required_fields(self) -> frozenset[str]:
-        return frozenset({"close"})
+        return frozenset({"close", "price_limit_pct"})
 
     def required_warmup_bars(self, params: dict) -> int:
         del params
         return 60
 
-    @staticmethod
-    def _limit_pct(market: MarketDataMatrix) -> np.ndarray:
-        values = np.full(len(market.symbols), 0.10, dtype=np.float32)
-        for asset_id, (symbol, name) in enumerate(zip(market.symbols, market.names, strict=True)):
-            if symbol.startswith(("300", "301", "688")):
-                values[asset_id] = 0.20
-            elif symbol.endswith(".BJ"):
-                values[asset_id] = 0.30
-            elif "ST" in name.upper():
-                values[asset_id] = 0.05
-        return values
-
     def compute_signals(self, market: MarketDataMatrix, params: dict) -> SignalMatrix:
         change = matrix_feature(market, "change_pct")
         entry = np.ones(market.shape, dtype=bool)
         if params.get("use_change_filter", True):
             entry &= change > float(params.get("min_change", 7.0)) / 100.0
         if params.get("use_limit_gap_filter", True):
+            limit_pct = matrix_feature(market, "price_limit_pct")
             entry &= (
                 change
-                < self._limit_pct(market)[None, :] - float(params.get("limit_gap", 3.0)) / 100.0
+                >= limit_pct - float(params.get("limit_gap", 3.0)) / 100.0
             )
         ma20 = matrix_feature(market, "ma20")
         exit_ = (market.close < ma20) & (shift(market.close, 1) >= shift(ma20, 1))

backend/app/tickflow/repository.py

@@ -75,7 +75,7 @@ class DataStore:
             (self.data_dir / sub).mkdir(parents=True, exist_ok=True)
 
         # 财务数据子目录
-        for sub in ("metrics", "income", "balance_sheet", "cash_flow"):
+        for sub in ("metrics", "income", "balance_sheet", "cash_flow", "shares"):
             (self.data_dir / "financials" / sub).mkdir(parents=True, exist_ok=True)
 
         # DuckDB 内存模式 — 不建 .db 文件(§7.1)
@@ -183,6 +183,8 @@ class DataStore:
                 SELECT * FROM read_parquet('{d}/financials/balance_sheet/*.parquet', union_by_name=true)""",
             f"""CREATE OR REPLACE VIEW financials_cash_flow AS
                 SELECT * FROM read_parquet('{d}/financials/cash_flow/*.parquet', union_by_name=true)""",
+            f"""CREATE OR REPLACE VIEW financials_shares AS
+                SELECT * FROM read_parquet('{d}/financials/shares/*.parquet', union_by_name=true)""",
             # 五档盘口 sealed 真假涨停(独立旁路存储,不进 enriched)
             f"""CREATE OR REPLACE VIEW depth5 AS
                 SELECT * FROM read_parquet('{d}/depth5/**/*.parquet', union_by_name=true)""",
@@ -300,6 +302,8 @@ class KlineRepository:
         self._live_agg_cache_date: date | None = None
         self._live_agg_check_date: date | None = None          # 上次跨日校验时的 today (快路径节流)
         self._instruments_cache: pl.DataFrame | None = None
+        self._historical_shares_cache: pl.DataFrame | None = None
+        self._historical_shares_mtime_ns: int | None = None
         # 完整 enriched 历史 (含所有指标, 供 filter_history 策略使用)
         self._enriched_history_cache: pl.DataFrame | None = None  # ~100万行
         self._enriched_history_start: date | None = None
@@ -544,7 +548,11 @@ class KlineRepository:
                     if instruments is not None and not instruments.is_empty():
                         step = time.perf_counter()
                         logger.info("enriched refresh step start: compute limit signals")
-                        df_full = compute_limit_signals(df_full, instruments)
+                        df_full = compute_limit_signals(
+                            df_full,
+                            instruments,
+                            historical_shares=self.get_historical_shares(),
+                        )
                         logger.info("enriched refresh step done: compute limit signals (%.2fs)", time.perf_counter() - step)
 
                     # JOIN instruments 到完整历史 (filter_history/basic_filter 需要 name/股本等列)
@@ -1052,6 +1060,16 @@ class KlineRepository:
             return pl.DataFrame()
         return self._instruments_cache
 
+    def get_historical_shares(self) -> pl.DataFrame:
+        """读取财务股本历史，并在文件更新后自动刷新缓存。"""
+        path = self.store.data_dir / "financials" / "shares" / "part.parquet"
+        mtime_ns = path.stat().st_mtime_ns if path.exists() else None
+        if self._historical_shares_cache is None or mtime_ns != self._historical_shares_mtime_ns:
+            from app.share_capital import load_share_history
+            self._historical_shares_cache = load_share_history(self.store.data_dir)
+            self._historical_shares_mtime_ns = mtime_ns
+        return self._historical_shares_cache
+
     def get_index_instruments(self) -> pl.DataFrame:
         """返回缓存的指数 instruments DataFrame。如无缓存则懒加载。"""
         if self._index_instruments_cache is None:
@@ -1380,7 +1398,11 @@ class KlineRepository:
             df = compute_indicators(df)
             df = compute_signals(df)
             instruments = self.get_instruments()
-            df = compute_limit_signals(df, instruments)
+            df = compute_limit_signals(
+                df,
+                instruments,
+                historical_shares=self.get_historical_shares(),
+            )
         except Exception as e:  # noqa: BLE001
             logger.warning("on-demand compute failed: %s", e)
         return df

backend/tests/backtest/test_matrix_strategy.py

@@ -597,7 +597,7 @@ def test_matrix_cache_prunes_by_bytes_and_leaves_no_staging_directory(tmp_path):
     del first
     gc.collect()
     assert second.close[0, 0] == pytest.approx(11.0)
-    assert len(list(cache_root.glob("v3-*"))) == 1
+    assert len(list(cache_root.glob("v4-*"))) == 1
     assert list(cache_root.glob(".*.tmp")) == []
     assert len(list(cache_root.glob(".axes-v1-*.json"))) == 1
 
@@ -650,7 +650,7 @@ def test_managed_source_generation_skips_file_walk_and_invalidates_explicitly(tm
     assert changed.cache_path != first.cache_path
     del first, repeated
     gc.collect()
-    assert len(list(cache_root.glob("v3-*"))) == 1
+    assert len(list(cache_root.glob("v4-*"))) == 1
 
 
 def test_registered_builtin_matrix_strategies_share_one_cache_profile():

backend/tests/test_financial_shares.py

@@ -0,0 +1,161 @@
+from __future__ import annotations
+
+from datetime import date
+
+import polars as pl
+import pytest
+
+from app.indicators import pipeline
+from app.services import financial_sync
+from app.tickflow.capabilities import CapabilitySet
+
+
+def _write_instruments(data_dir, symbols: list[str]) -> None:
+    path = data_dir / "instruments" / "instruments.parquet"
+    path.parent.mkdir(parents=True, exist_ok=True)
+    pl.DataFrame({"symbol": symbols}).write_parquet(path)
+
+
+def test_first_share_sync_fetches_complete_history(tmp_path, monkeypatch):
+    _write_instruments(tmp_path, ["600000.SH"])
+    calls: list[tuple[list[str], bool]] = []
+
+    def fake_fetch(table, symbols, capset, latest_only=True):
+        assert table == "shares"
+        calls.append((symbols, latest_only))
+        return pl.DataFrame({
+            "symbol": ["600000.SH", "600000.SH"],
+            "period_end": ["2023-12-31", "2024-06-30"],
+            "float_shares": [10.0, 12.0],
+        })
+
+    monkeypatch.setattr(financial_sync, "_fetch_table", fake_fetch)
+
+    rows = financial_sync.sync_shares(tmp_path, CapabilitySet())
+
+    assert rows == 2
+    assert calls == [(["600000.SH"], False)]
+    stored = pl.read_parquet(tmp_path / "financials" / "shares" / "part.parquet")
+    assert stored["period_end"].to_list() == ["2023-12-31", "2024-06-30"]
+
+
+def test_incremental_share_sync_updates_existing_and_backfills_new_symbols(tmp_path, monkeypatch):
+    _write_instruments(tmp_path, ["600000.SH", "000001.SZ"])
+    path = tmp_path / "financials" / "shares" / "part.parquet"
+    path.parent.mkdir(parents=True, exist_ok=True)
+    pl.DataFrame({
+        "symbol": ["600000.SH"],
+        "period_end": ["2024-06-30"],
+        "float_shares": [10.0],
+    }).write_parquet(path)
+    calls: list[tuple[list[str], bool]] = []
+
+    def fake_fetch(table, symbols, capset, latest_only=True):
+        assert table == "shares"
+        calls.append((symbols, latest_only))
+        if latest_only:
+            return pl.DataFrame({
+                "symbol": ["600000.SH"],
+                "period_end": ["2024-06-30"],
+                "float_shares": [11.0],
+            })
+        return pl.DataFrame({
+            "symbol": ["000001.SZ", "000001.SZ"],
+            "period_end": ["2023-12-31", "2024-06-30"],
+            "float_shares": [20.0, 21.0],
+        })
+
+    monkeypatch.setattr(financial_sync, "_fetch_table", fake_fetch)
+
+    rows = financial_sync.sync_shares(tmp_path, CapabilitySet())
+
+    assert rows == 3
+    assert calls == [(["000001.SZ"], False), (["600000.SH"], True)]
+    stored = pl.read_parquet(path).sort(["symbol", "period_end"])
+    assert stored.filter(pl.col("symbol") == "600000.SH")["float_shares"].to_list() == [11.0]
+    assert stored.filter(pl.col("symbol") == "000001.SZ")["float_shares"].to_list() == [20.0, 21.0]
+
+
+def test_custom_financial_provider_receives_shares_contract(monkeypatch):
+    received: list[tuple[str, list[str], bool]] = []
+
+    class Provider:
+        def get_financials(self, table, symbols, latest_only=True):
+            received.append((table, symbols, latest_only))
+            return pl.DataFrame({
+                "symbol": symbols,
+                "period_end": ["2024-06-30"],
+                "float_shares": [10.0],
+            })
+
+    from app.data_providers import custom as custom_sources
+    from app.services import preferences
+
+    monkeypatch.setattr(financial_sync, "_financial_is_custom", lambda: True)
+    monkeypatch.setattr(preferences, "get_financial_provider", lambda: "custom-test")
+    monkeypatch.setattr(custom_sources, "get_provider", lambda _name: Provider())
+
+    result = financial_sync._fetch_table(
+        "shares",
+        ["600000.SH"],
+        CapabilitySet(),
+        latest_only=False,
+    )
+
+    assert result.height == 1
+    assert received == [("shares", ["600000.SH"], False)]
+
+
+def test_historical_turnover_uses_only_available_share_capital(monkeypatch):
+    monkeypatch.setattr(pipeline, "cn_today", lambda: date(2026, 7, 18))
+    bars = pl.DataFrame({
+        "symbol": ["600000.SH"] * 5,
+        "date": [
+            date(2024, 3, 31),
+            date(2024, 4, 14),
+            date(2024, 4, 15),
+            date(2024, 6, 30),
+            date(2026, 7, 18),
+        ],
+        "volume": [10_000.0] * 5,
+    })
+    instruments = pl.DataFrame({
+        "symbol": ["600000.SH"],
+        "float_shares": [200_000_000.0],
+    })
+    shares = pl.DataFrame({
+        "symbol": ["600000.SH", "600000.SH"],
+        "period_end": ["2023-12-31", "2024-06-30"],
+        "announce_date": ["2024-04-15", None],
+        "float_shares": [100_000_000.0, 50_000_000.0],
+    })
+
+    result = pipeline.compute_limit_signals(
+        bars,
+        instruments,
+        needed={"turnover_rate"},
+        historical_shares=shares,
+    )
+
+    assert result["turnover_rate"].to_list() == pytest.approx([0.5, 0.5, 1.0, 2.0, 0.5])
+
+
+def test_turnover_without_share_history_keeps_existing_behavior(monkeypatch):
+    monkeypatch.setattr(pipeline, "cn_today", lambda: date(2026, 7, 18))
+    bars = pl.DataFrame({
+        "symbol": ["600000.SH"],
+        "date": [date(2024, 4, 15)],
+        "volume": [10_000.0],
+    })
+    instruments = pl.DataFrame({
+        "symbol": ["600000.SH"],
+        "float_shares": [200_000_000.0],
+    })
+
+    result = pipeline.compute_limit_signals(
+        bars,
+        instruments,
+        needed={"turnover_rate"},
+    )
+
+    assert result["turnover_rate"][0] == pytest.approx(0.5)

backend/tests/test_price_limits.py

@@ -0,0 +1,147 @@
+from __future__ import annotations
+
+from datetime import date
+
+import numpy as np
+import polars as pl
+import pytest
+
+from app.api import kline
+from app.backtest.matrix import load_market_data_matrix_from_parquet
+from app.price_limits import (
+    numpy_limit_price,
+    numpy_price_limit_matrix,
+    polars_is_risk_warning_name,
+    polars_limit_price,
+    polars_price_limit_pct,
+    price_limit_pct,
+)
+
+
+@pytest.mark.parametrize(
+    ("symbol", "trade_date", "is_st", "expected"),
+    [
+        ("600001.SH", date(2026, 7, 3), True, 0.05),
+        ("600001.SH", date(2026, 7, 6), True, 0.10),
+        ("000001.SZ", date(2026, 7, 3), False, 0.10),
+        ("300001.SZ", date(2026, 7, 3), True, 0.20),
+        ("688001.SH", date(2026, 7, 3), True, 0.20),
+        ("689001.SH", date(2026, 7, 3), True, 0.20),
+        ("830001.BJ", date(2026, 7, 3), True, 0.30),
+    ],
+)
+def test_scalar_price_limit_rules(symbol, trade_date, is_st, expected):
+    assert price_limit_pct(
+        symbol,
+        trade_date,
+        is_risk_warning=is_st,
+    ) == pytest.approx(expected)
+
+
+def test_polars_and_numpy_price_limit_rules_match():
+    dates = [date(2026, 7, 3), date(2026, 7, 6)]
+    symbols = ["600001.SH", "300001.SZ", "689001.SH", "830001.BJ"]
+    names = ["*st主板", "*ST创业", "科创ST", "北交ST"]
+    panel = pl.DataFrame({
+        "date": [value for value in dates for _ in symbols],
+        "symbol": symbols * len(dates),
+        "name": names * len(dates),
+    }).with_columns(
+        polars_is_risk_warning_name(pl.col("name")).alias("is_st")
+    ).with_columns(
+        polars_price_limit_pct(
+            pl.col("symbol"), pl.col("date"), pl.col("is_st"),
+        ).alias("limit_pct")
+    )
+    polars_values = panel["limit_pct"].to_numpy().reshape(len(dates), len(symbols))
+    numpy_values = numpy_price_limit_matrix(dates, symbols, names)
+    np.testing.assert_allclose(polars_values, numpy_values)
+
+
+def test_polars_and_numpy_limit_prices_use_identical_half_up_rounding():
+    previous = np.array([18.90, 10.00], dtype=np.float64)
+    limits = np.array([0.05, 0.10], dtype=np.float64)
+    frame = pl.DataFrame({"previous": previous, "limit": limits})
+
+    for up in (True, False):
+        polars_values = frame.select(
+            polars_limit_price(
+                pl.col("previous"), pl.col("limit"), up=up,
+            ).alias("price")
+        )["price"].to_numpy()
+        numpy_values = numpy_limit_price(previous, limits, up=up)
+        np.testing.assert_allclose(polars_values, numpy_values)
+    assert numpy_limit_price(previous, limits, up=False)[0] == pytest.approx(17.96)
+
+
+def test_matrix_uses_date_specific_st_limits_across_change(tmp_path):
+    root = tmp_path / "market"
+    rows = [
+        (date(2026, 7, 2), 10.0),
+        (date(2026, 7, 3), 10.5),
+        (date(2026, 7, 6), 11.03),
+    ]
+    for trade_date, close in rows:
+        partition = root / f"date={trade_date.isoformat()}"
+        partition.mkdir(parents=True)
+        pl.DataFrame({
+            "symbol": ["600001.SH"],
+            "date": [trade_date],
+            "open": [close],
+            "high": [close],
+            "low": [close],
+            "close": [close],
+            "raw_close": [close],
+            "volume": [1000.0],
+        }).write_parquet(partition / "part.parquet")
+
+    market = load_market_data_matrix_from_parquet(
+        root,
+        rows[0][0],
+        rows[-1][0],
+        field_columns={"raw_close", "price_limit_pct"},
+        instruments=pl.DataFrame({
+            "symbol": ["600001.SH"],
+            "name": ["*ST主板"],
+        }),
+        cache_root=tmp_path / "cache",
+    )
+    np.testing.assert_allclose(
+        market.field("price_limit_pct")[:, 0],
+        np.array([0.05, 0.05, 0.10], dtype=np.float32),
+    )
+    assert market.limit_up_locked[:, 0].tolist() == [0, 1, 0]
+
+
+class _InstrumentRepo:
+    def get_instruments_asset(self, asset_type: str) -> pl.DataFrame:
+        assert asset_type == "stock"
+        return pl.DataFrame({
+            "symbol": ["600001.SH"],
+            "limit_up": [10.88],
+            "limit_down": [8.90],
+        })
+
+
+def test_minute_price_limit_prefers_authoritative_prices_only_today(monkeypatch):
+    today = date(2026, 7, 18)
+    monkeypatch.setattr(kline, "cn_today", lambda: today)
+    current = kline._get_price_limit_info(
+        _InstrumentRepo(), "600001.SH", today, "stock", "*ST主板",
+    )
+    historical = kline._get_price_limit_info(
+        _InstrumentRepo(), "600001.SH", date(2026, 7, 3), "stock", "*ST主板",
+    )
+
+    assert current == {
+        "rate": 0.10,
+        "limit_up": 10.88,
+        "limit_down": 8.90,
+        "source": "instrument",
+    }
+    assert historical == {
+        "rate": 0.05,
+        "limit_up": None,
+        "limit_down": None,
+        "source": "rule",
+    }

backend/tests/test_st_limit_and_sharpe.py

@@ -1,13 +1,14 @@
 """回归测试:
 
-1. ST 5% 涨跌停限幅仅适用于主板风险警示股; 创业板/科创板 ST 仍执行 20%。
+1. 2026-07-06 前 ST 5% 涨跌停限幅仅适用于主板风险警示股;
+   新规生效后主板 ST 为 10%, 创业板/科创板 ST 始终执行 20%。
    (修正前 _is_st 无条件套 5%, 会误报/漏报这批股的涨停。)
 2. 因子回测 Sharpe 的年化系数须匹配调仓频率 (月频 √12 / 周频 √52 / 日频 √252);
    (修正前一律 √252, 月频 Sharpe 被高估 √21 ≈ 4.6 倍。)
 """
 from __future__ import annotations
 
-from datetime import date
+from datetime import date, timedelta
 
 import polars as pl
 import pytest
@@ -20,29 +21,37 @@ from app.strategy.builtin.near_limit_up import MATRIX_STRATEGY
 
 def test_near_limit_pct_st_only_on_main_board():
     df = pl.DataFrame({
-        "symbol": ["300001", "688001", "600001", "000001", "830001.BJ"],
-        "name": ["*ST创业", "科创ST", "*ST主板", "平安银行", "北交ST"],
-        "date": [date(2024, 1, 2)] * 5,
-        "open": [10.0] * 5,
-        "high": [10.0] * 5,
-        "low": [10.0] * 5,
-        "close": [10.0] * 5,
-        "volume": [1000.0] * 5,
+        "symbol": ["300001", "688001", "689001", "600001", "000001", "830001.BJ"],
+        "name": ["*ST创业", "科创ST", "科创ST", "*ST主板", "平安银行", "北交ST"],
+        "date": [date(2024, 1, 2)] * 6,
+        "open": [10.0] * 6,
+        "high": [10.0] * 6,
+        "low": [10.0] * 6,
+        "close": [10.0] * 6,
+        "volume": [1000.0] * 6,
     })
-    market = build_market_data_matrix(df)
-    limit_by_symbol = dict(zip(market.symbols, MATRIX_STRATEGY._limit_pct(market), strict=True))
+    market = build_market_data_matrix(df, field_columns={"price_limit_pct"})
+    limit_by_symbol = dict(
+        zip(market.symbols, market.field("price_limit_pct")[0], strict=True)
+    )
     assert limit_by_symbol["300001"] == pytest.approx(0.20)  # 创业板 ST → 20%
     assert limit_by_symbol["688001"] == pytest.approx(0.20)  # 科创板 ST → 20%
+    assert limit_by_symbol["689001"] == pytest.approx(0.20)  # 科创板 689 → 20%
     assert limit_by_symbol["600001"] == pytest.approx(0.05)  # 主板 ST → 5%
     assert limit_by_symbol["000001"] == pytest.approx(0.10)  # 主板普通 → 10%
     assert limit_by_symbol["830001.BJ"] == pytest.approx(0.30)  # 北交所 → 30%
 
 
-def _two_day(symbol: str, prev_close: float, today_close: float) -> pl.DataFrame:
+def _two_day(
+    symbol: str,
+    prev_close: float,
+    today_close: float,
+    trade_date: date = date(2024, 1, 3),
+) -> pl.DataFrame:
     """2 日最小输入: 首日平收, 次日收于 today_close。"""
     return pl.DataFrame({
         "symbol": [symbol, symbol],
-        "date": [date(2024, 1, 2), date(2024, 1, 3)],
+        "date": [trade_date - timedelta(days=1), trade_date],
         "raw_close": [prev_close, today_close],
         "close": [prev_close, today_close],
         "raw_high": [prev_close, today_close],
@@ -54,8 +63,14 @@ def _two_day(symbol: str, prev_close: float, today_close: float) -> pl.DataFrame
     })
 
 
-def _last_limit_up(symbol: str, name: str, prev_close: float, today_close: float):
-    df = _two_day(symbol, prev_close, today_close)
+def _last_limit_up(
+    symbol: str,
+    name: str,
+    prev_close: float,
+    today_close: float,
+    trade_date: date = date(2024, 1, 3),
+):
+    df = _two_day(symbol, prev_close, today_close, trade_date)
     inst = pl.DataFrame({"symbol": [symbol], "name": [name]})
     out = compute_limit_signals(df, inst).sort("date")
     return out["signal_limit_up"].to_list()[-1], out["consecutive_limit_ups"].to_list()[-1]
@@ -74,12 +89,45 @@ def test_st_chinext_plus5pct_is_not_a_false_limit_up():
     assert sig is False
 
 
-def test_st_main_board_still_limits_at_5pct():
-    # 主板 *ST 昨收 10.00 → 今日 +5% 至 10.50 仍应识别为涨停
+def test_st_main_board_historical_limit_is_5pct():
+    # 新规前主板 *ST 昨收 10.00 → 今日 +5% 至 10.50 应识别为涨停
     sig, _ = _last_limit_up("600001", "*ST主板", 10.0, 10.5)
     assert sig is True
 
 
+def test_st_main_board_limit_changes_to_10pct_on_2026_07_06():
+    change_date = date(2026, 7, 6)
+    plus_five, _ = _last_limit_up(
+        "600001", "*ST主板", 10.0, 10.5, change_date,
+    )
+    plus_ten, _ = _last_limit_up(
+        "600001", "*ST主板", 10.0, 11.0, change_date,
+    )
+    assert plus_five is False
+    assert plus_ten is True
+
+
+def test_near_limit_up_accepts_stocks_inside_configured_gap():
+    dates = [date(2026, 6, 8) + timedelta(days=index) for index in range(21)]
+    closes = [10.0] * 20 + [10.8]
+    panel = pl.DataFrame({
+        "symbol": ["600001.SH"] * len(dates),
+        "name": ["普通股"] * len(dates),
+        "date": dates,
+        "open": closes,
+        "high": closes,
+        "low": closes,
+        "close": closes,
+        "volume": [1000.0] * len(dates),
+    })
+    market = build_market_data_matrix(panel, field_columns={"price_limit_pct"})
+    signals = MATRIX_STRATEGY.compute_signals(market, {
+        "min_change": 7.0,
+        "limit_gap": 3.0,
+    })
+    assert signals.entry[-1, 0] == 1
+
+
 def test_sharpe_annualization_matches_rebalance_frequency():
     nav = [
         {"date": "2024-01-31", "Q1": 1.00},

frontend/src/components/EChartsIntraday.tsx

@@ -1,7 +1,7 @@
 import { useEffect, useMemo, useRef, useState } from 'react'
 import * as echarts from 'echarts'
 import type { ECharts, EChartsOption } from 'echarts'
-import type { MinuteKlineRow } from '@/lib/api'
+import type { MinuteKlineRow, PriceLimitInfo } from '@/lib/api'
 import { useChartTheme, type ChartTheme } from '@/lib/theme'
 
 type YMode = 'adaptive' | 'limit'
@@ -20,7 +20,7 @@ interface Props {
   height?: number
   prevClose?: number
   date?: string
-  symbol?: string
+  priceLimit?: PriceLimitInfo
   onPriceHover?: (price: number | null) => void
   showLimitLines?: boolean
   showAvgLine?: boolean
@@ -79,34 +79,29 @@ function generateFullDayTimes(): string[] {
 
 const FULL_DAY_TIMES = generateFullDayTimes()
 
-/** 根据 symbol 判断涨跌停幅度 (创业板/科创板 ±20%, 北交所 ±30%, 其余 ±10%) */
-function getLimitPct(symbol?: string): number {
-  if (!symbol) return 0.10
-  if (symbol.endsWith('.BJ')) return 0.30                                  // 北交所
-  if (symbol.startsWith('300') || symbol.startsWith('301')) return 0.20  // 创业板
-  if (symbol.startsWith('688') || symbol.startsWith('689')) return 0.20  // 科创板
-  return 0.10
-}
-
 /** 计算实际涨跌停价 (四舍五入到2位小数) 和实际涨跌停幅度 */
-function getLimitPrices(prevClose: number, symbol?: string): {
+function getLimitPrices(prevClose: number, priceLimit?: PriceLimitInfo): {
   limitUp: number      // 涨停价 (四舍五入)
   limitDown: number    // 跌停价 (四舍五入)
   upPct: number        // 实际涨停幅度 (如 9.97)
   downPct: number      // 实际跌停幅度 (如 -9.97)
 } {
-  const pct = getLimitPct(symbol)
+  const pct = priceLimit && Number.isFinite(priceLimit.rate) ? priceLimit.rate : 0.10
   const rawUp = prevClose * (1 + pct)
   const rawDown = prevClose * (1 - pct)
   // A股涨跌停价四舍五入到分 (2位小数)
-  const limitUp = Math.round(rawUp * 100) / 100
-  const limitDown = Math.round(rawDown * 100) / 100
+  const limitUp = isValidPrice(priceLimit?.limit_up)
+    ? priceLimit.limit_up
+    : Math.round(rawUp * 100) / 100
+  const limitDown = isValidPrice(priceLimit?.limit_down)
+    ? priceLimit.limit_down
+    : Math.round(rawDown * 100) / 100
   const upPct = (limitUp - prevClose) / prevClose * 100
   const downPct = (limitDown - prevClose) / prevClose * 100
   return { limitUp, limitDown, upPct, downPct }
 }
 
-function buildOption(data: MinuteKlineRow[], prevClose: number | undefined, avgPrices: number[], lineColor: string, areaColor: string, yMode: YMode, ct: ChartTheme, symbol?: string, showLimitLines = true, showAvgLine = true): EChartsOption {
+function buildOption(data: MinuteKlineRow[], prevClose: number | undefined, avgPrices: number[], lineColor: string, areaColor: string, yMode: YMode, ct: ChartTheme, priceLimit?: PriceLimitInfo, showLimitLines = true, showAvgLine = true): EChartsOption {
   // 将数据映射到全天时间轴上的正确位置
   const timeIndexMap = new Map(FULL_DAY_TIMES.map((t, i) => [t, i]))
   const closes = new Array(FULL_DAY_TIMES.length).fill(null) as (number | null)[]
@@ -168,7 +163,7 @@ function buildOption(data: MinuteKlineRow[], prevClose: number | undefined, avgP
     }
 
     if (showLimitLines && yMode === 'limit') {
-      const { limitUp, limitDown } = getLimitPrices(prevClose, symbol)
+      const { limitUp, limitDown } = getLimitPrices(prevClose, priceLimit)
       const limitDiffUp = limitUp - prevClose
       const limitDiffDown = prevClose - limitDown
       const limitDiff = Math.max(limitDiffUp, limitDiffDown)
@@ -194,7 +189,7 @@ function buildOption(data: MinuteKlineRow[], prevClose: number | undefined, avgP
     } else {
       // 自适应模式: Y 轴按实际涨跌幅对称, 但不超出实际涨跌停范围
       if (showLimitLines) {
-        const { limitUp, limitDown } = getLimitPrices(prevClose, symbol)
+        const { limitUp, limitDown } = getLimitPrices(prevClose, priceLimit)
         const limitDiff = Math.max(limitUp - prevClose, prevClose - limitDown)
         maxDiff = Math.min(maxDiff, limitDiff)
       }
@@ -404,7 +399,7 @@ function buildOption(data: MinuteKlineRow[], prevClose: number | undefined, avgP
   }
 }
 
-export function EChartsIntraday({ data, height = 320, prevClose, date, symbol, onPriceHover, showLimitLines = true, showAvgLine = true }: Props) {
+export function EChartsIntraday({ data, height = 320, prevClose, date, priceLimit, onPriceHover, showLimitLines = true, showAvgLine = true }: Props) {
   const containerRef = useRef<HTMLDivElement>(null)
   const chartRef = useRef<ECharts | null>(null)
   const roRef = useRef<ResizeObserver | null>(null)
@@ -493,11 +488,11 @@ export function EChartsIntraday({ data, height = 320, prevClose, date, symbol, o
       }
       fullDayToDataIdx.current = mapping
 
-      chart.setOption(buildOption(data, prevClose, avgPrices, lineColor, areaFill, yMode, ct, symbol, showLimitLines, showAvgLine), true)
+      chart.setOption(buildOption(data, prevClose, avgPrices, lineColor, areaFill, yMode, ct, priceLimit, showLimitLines, showAvgLine), true)
     } else {
       chart.clear()
     }
-  }, [data, prevClose, height, lineColor, areaFill, yMode, ct, symbol, showLimitLines, showAvgLine])
+  }, [data, prevClose, height, lineColor, areaFill, yMode, ct, priceLimit, showLimitLines, showAvgLine])
 
   useEffect(() => {
     return () => {

frontend/src/components/StockIntradayChart.tsx

@@ -114,7 +114,7 @@ export function StockIntradayChart({
           height={height}
           prevClose={prevClose}
           date={date}
-          symbol={symbol}
+          priceLimit={minute.data?.price_limit ?? undefined}
           onPriceHover={onPriceHover}
         />
       )}

frontend/src/components/financials/StockFinancialDetail.tsx

@@ -1,11 +1,12 @@
 import { useState } from 'react'
 import { motion, AnimatePresence } from 'framer-motion'
-import { CalendarDays, TrendingUp, FileText, Wallet, Activity, Sparkles, AlertTriangle, Loader2 } from 'lucide-react'
+import { CalendarDays, TrendingUp, FileText, Wallet, Activity, Sparkles, AlertTriangle, Loader2, ChartPie } from 'lucide-react'
 import {
   useFinancialMetrics,
   useFinancialIncome,
   useFinancialBalanceSheet,
   useFinancialCashFlow,
+  useFinancialShares,
 } from '@/lib/useFinancials'
 import { fmtPrice, fmtBigNum, fmtDate } from '@/lib/format'
 import { Skeleton } from '@/components/data/Skeleton'
@@ -17,13 +18,14 @@ interface Props {
   name: string
 }
 
-type TabKey = 'metrics' | 'income' | 'balance_sheet' | 'cash_flow'
+type TabKey = 'metrics' | 'income' | 'balance_sheet' | 'cash_flow' | 'shares'
 
 const TABS: { key: TabKey; label: string; icon: typeof TrendingUp }[] = [
   { key: 'metrics', label: '核心指标', icon: TrendingUp },
   { key: 'income', label: '利润表', icon: FileText },
   { key: 'balance_sheet', label: '资产负债表', icon: Wallet },
   { key: 'cash_flow', label: '现金流量表', icon: Activity },
+  { key: 'shares', label: '股本', icon: ChartPie },
 ]
 
 // 字段定义:键 → (中文名, 格式化类型)
@@ -94,6 +96,10 @@ const FIELD_DEFS: Record<TabKey, FieldDef[]> = {
     { label: '固定资产/无形资产投资', fmt: 'amount', key: 'capex' } as any,
     { label: '现金及等价物净增加额', fmt: 'amount', key: 'net_cash_change' } as any,
   ],
+  shares: [
+    { label: '总股本', fmt: 'amount', key: 'total_shares' } as any,
+    { label: '流通股本', fmt: 'amount', key: 'float_shares' } as any,
+  ],
 }
 
 function formatValue(v: number | null | undefined, fmt: FmtType): string {
@@ -148,12 +154,14 @@ export function StockFinancialDetail({ symbol, name }: Props) {
   const income = useFinancialIncome(symbol)
   const balance = useFinancialBalanceSheet(symbol)
   const cashFlow = useFinancialCashFlow(symbol)
+  const shares = useFinancialShares(symbol)
 
   const queryMap = {
     metrics: metrics,
     income: income,
     balance_sheet: balance,
     cash_flow: cashFlow,
+    shares: shares,
   } as const
 
   const current = queryMap[tab]
@@ -198,7 +206,7 @@ export function StockFinancialDetail({ symbol, name }: Props) {
       </div>
 
       {/* 标签页 */}
-      <div className="flex items-center gap-1 px-3 pt-2 border-b border-border/60">
+      <div className="flex items-center gap-1 px-3 pt-2 border-b border-border/60 overflow-x-auto">
         {TABS.map(t => {
           const Icon = t.icon
           const isActive = tab === t.key

frontend/src/lib/api.ts

@@ -118,6 +118,15 @@ export interface FinancialCashFlowRecord {
   [key: string]: any
 }
 
+export interface FinancialSharesRecord {
+  symbol?: string
+  period_end: string
+  announce_date?: string | null
+  total_shares?: number | null
+  float_shares?: number | null
+  [key: string]: any
+}
+
 /** AI 财务分析历史报告 */
 export interface AiFinancialReport {
   id: string
@@ -185,6 +194,13 @@ export interface MinuteKlineRow {
   amount: number
 }
 
+export interface PriceLimitInfo {
+  rate: number
+  limit_up: number | null
+  limit_down: number | null
+  source: 'rule' | 'instrument'
+}
+
 export interface KlineRow {
   symbol?: string
   date: string
@@ -1250,6 +1266,7 @@ export const api = {
       date: string | null
       rows: MinuteKlineRow[]
       source?: 'local' | 'live' | 'none'
+      price_limit?: PriceLimitInfo | null
     }>(
       `/api/kline/minute?symbol=${encodeURIComponent(symbol)}${date ? `&date=${date}` : ''}`,
     ),
@@ -1665,6 +1682,11 @@ export const api = {
       `/api/financials/cash-flow${symbol ? `?symbol=${encodeURIComponent(symbol)}` : ''}`,
     ),
 
+  financialShares: (symbol?: string) =>
+    request<{ data: FinancialSharesRecord[] }>(
+      `/api/financials/shares${symbol ? `?symbol=${encodeURIComponent(symbol)}` : ''}`,
+    ),
+
   /** 触发财务数据同步(后台异步执行,接口立即返回 started 状态) */
   financialSync: (table: string) =>
     request<{ status: string; synced: { started: boolean; reason?: string } }>(

frontend/src/lib/useFinancials.ts

@@ -7,6 +7,7 @@ export const FINANCIAL_QK = {
   income: (symbol?: string) => ['financials', 'income', symbol],
   balanceSheet: (symbol?: string) => ['financials', 'balance-sheet', symbol],
   cashFlow: (symbol?: string) => ['financials', 'cash-flow', symbol],
+  shares: (symbol?: string) => ['financials', 'shares', symbol],
 }
 
 export function useFinancialStatus() {
@@ -55,6 +56,15 @@ export function useFinancialCashFlow(symbol?: string) {
   })
 }
 
+export function useFinancialShares(symbol?: string) {
+  return useQuery({
+    queryKey: FINANCIAL_QK.shares(symbol),
+    queryFn: () => api.financialShares(symbol),
+    enabled: !!symbol,
+    staleTime: 300_000,
+  })
+}
+
 export function useFinancialSync() {
   const qc = useQueryClient()
   return useMutation({

frontend/src/pages/Financials.tsx

@@ -1,5 +1,5 @@
 import { useState, useEffect } from 'react'
-import { RefreshCw, Lock, Loader2, X, Search, FileText, Database, Clock, CheckCircle2, Hourglass, Lightbulb, ExternalLink } from 'lucide-react'
+import { RefreshCw, Lock, Loader2, X, Search, FileText, Database, Clock, CheckCircle2, Hourglass, Lightbulb, ExternalLink, ChartPie } from 'lucide-react'
 import { PageHeader } from '@/components/PageHeader'
 import { EmptyState } from '@/components/EmptyState'
 import { useCapabilities } from '@/lib/useSharedQueries'
@@ -17,6 +17,7 @@ const TABLE_LABELS: Record<string, string> = {
   income: '利润表',
   balance_sheet: '资产负债表',
   cash_flow: '现金流量表',
+  shares: '股本表',
 }
 
 const TABLE_ICON: Record<string, typeof FileText> = {
@@ -24,12 +25,13 @@ const TABLE_ICON: Record<string, typeof FileText> = {
   income: FileText,
   balance_sheet: FileText,
   cash_flow: FileText,
+  shares: ChartPie,
 }
 
 export function Financials() {
   const { data: caps } = useCapabilities()
-  const hasFinancial = caps?.capabilities?.['financial'] != null
   const { data: status, isLoading } = useFinancialStatus()
+  const hasFinancial = caps?.capabilities?.['financial'] != null || status?.available === true
   const syncMut = useFinancialSync()
   // 同步进行中 = 服务端真值(status.syncing)或本地乐观态(请求已发出待确认)。
   // 乐观窗口:点击后到 invalidate 触发的 refetch 返回之间,status.syncing 暂为 false,
@@ -60,7 +62,7 @@ export function Financials() {
   if (!hasFinancial) {
     return (
       <>
-        <PageHeader title="财务分析" subtitle="利润表 / 资负表 / 现金流 / 关键指标 / AI分析 · Expert" />
+        <PageHeader title="财务分析" subtitle="利润表 / 资负表 / 现金流 / 关键指标 / 股本 / AI分析 · Expert" />
         <div className="px-8 py-10">
           <div className="mx-auto max-w-md rounded-card border border-warning/30 bg-warning/[0.04] p-8 text-center">
             <div className="mx-auto flex h-12 w-12 items-center justify-center rounded-full bg-warning/10">
@@ -98,7 +100,7 @@ export function Financials() {
   const handleSync = (table: string) => {
     // 防重复点击:syncing 中不再触发(后端 trigger 也有 _is_syncing 兜底)
     if (syncing) return
-    // 记录开始时间: 全量同步判断所有 4 张表, 单表同步只判断这一张
+    // 记录开始时间: 全量同步判断所有财务表, 单表同步只判断这一张
     setSyncStartedAt(Date.now())
     setSyncSingleTable(table === 'all' ? null : table)
     syncMut.mutate(table, {
@@ -132,7 +134,7 @@ export function Financials() {
   // 本次同步进度: 仅当 syncStartedAt 存在且 syncing 时, 按 last_sync 时间戳判断
   const isFullSync = syncing && syncStartedAt && !syncSingleTable  // 全量同步
   const isSingleSync = syncing && syncStartedAt && !!syncSingleTable  // 单表同步
-  const TABLE_ORDER = ['metrics', 'income', 'balance_sheet', 'cash_flow'] as const
+  const TABLE_ORDER = ['metrics', 'income', 'balance_sheet', 'cash_flow', 'shares'] as const
   const tableDoneThisRound = (key: string): boolean => {
     if (!syncStartedAt || !syncing) return false
     // 单表同步: 只判断这一张表是否完成
@@ -157,7 +159,7 @@ export function Financials() {
     <>
       <PageHeader
         title="财务分析"
-        subtitle="利润表 / 资负表 / 现金流 / 关键指标 / AI分析 · Expert"
+        subtitle="利润表 / 资负表 / 现金流 / 关键指标 / 股本 / AI分析 · Expert"
         right={
           <div className="flex items-center gap-2">
             <LastStockChip stock={lastStock} onSelect={pick} />
@@ -165,7 +167,7 @@ export function Financials() {
               <span className="text-xs text-accent/80 flex items-center gap-1.5">
                 <Loader2 className="w-3 h-3 animate-spin" />
                 {isFullSync
-                  ? `已同步 ${syncedCount}/4 张表…`
+                  ? `已同步 ${syncedCount}/${TABLE_ORDER.length} 张表…`
                   : isSingleSync
                     ? `同步${TABLE_LABELS[syncSingleTable!] ?? syncSingleTable}…`
                     : '同步中…'}
@@ -186,18 +188,18 @@ export function Financials() {
         }
       />
 
-      <div className="px-8 py-6 space-y-6 max-w-7xl">
+      <div className="px-3 sm:px-8 py-6 space-y-6 max-w-7xl">
         {syncing && (
           <div className="flex items-center gap-2 rounded-card border border-accent/30 bg-accent/[0.06] px-3 py-2 text-xs text-accent">
             <Loader2 className="h-3.5 w-3.5 animate-spin shrink-0" />
-            正在从 TickFlow 拉取财务数据，请稍候…
+            正在从财务数据源拉取数据，请稍候…
           </div>
         )}
 
         {/* 同步状态卡片 —— 始终显示,反映本地财务数据概况 */}
         {!isLoading && available && (
           <div>
-            <div className="grid grid-cols-2 md:grid-cols-4 gap-3">
+            <div className="grid grid-cols-1 sm:grid-cols-2 md:grid-cols-3 xl:grid-cols-5 gap-3">
               {Object.entries(TABLE_LABELS).map(([key, label]) => {
                 const info = tables[key]
                 const TIcon = TABLE_ICON[key] ?? Database
@@ -316,7 +318,7 @@ export function Financials() {
                 <EmptyState
                   icon={Search}
                   title="未选择股票"
-                  hint="在上方搜索框输入股票代码或名称，选择后即可查看该股的核心指标、利润表、资产负债表与现金流量表。"
+                  hint="在上方搜索框输入股票代码或名称，选择后即可查看该股的核心指标、财务报表与股本历史。"
                 />
               )}
             </div>

frontend/src/pages/Indices.tsx

@@ -332,7 +332,6 @@ export function Indices() {
                         height={620}
                         prevClose={prevClose}
                         date={selectedDate ?? undefined}
-                        symbol={selectedSymbol}
                         showLimitLines={false}
                         showAvgLine={false}
                         onPriceHover={setLinkedPrice}