From 8e4258cf20cae40002cd5449426a23753e693b1c Mon Sep 17 00:00:00 2001 From: shy3130 <415333856@qq.com> Date: Sat, 18 Jul 2026 23:37:54 +0800 Subject: [PATCH] feat: support historical price limits and share capital --- backend/app/api/financials.py | 24 ++- backend/app/api/kline.py | 73 +++++++- backend/app/backtest/engine.py | 5 + backend/app/backtest/matrix.py | 84 +++++---- backend/app/data_providers/custom/provider.py | 11 +- backend/app/indicators/pipeline.py | 139 +++++++-------- backend/app/price_limits.py | 158 +++++++++++++++++ backend/app/services/financial_analyzer.py | 10 +- backend/app/services/financial_sync.py | 129 +++++++++++--- backend/app/services/quote_service.py | 11 +- backend/app/services/screener.py | 12 +- backend/app/share_capital.py | 110 ++++++++++++ backend/app/strategy/builtin/near_limit_up.py | 17 +- backend/app/tickflow/repository.py | 28 ++- .../tests/backtest/test_matrix_strategy.py | 4 +- backend/tests/test_financial_shares.py | 161 ++++++++++++++++++ backend/tests/test_price_limits.py | 147 ++++++++++++++++ backend/tests/test_st_limit_and_sharpe.py | 84 +++++++-- frontend/src/components/EChartsIntraday.tsx | 37 ++-- .../src/components/StockIntradayChart.tsx | 2 +- .../financials/StockFinancialDetail.tsx | 14 +- frontend/src/lib/api.ts | 22 +++ frontend/src/lib/useFinancials.ts | 10 ++ frontend/src/pages/Financials.tsx | 24 +-- frontend/src/pages/Indices.tsx | 1 - 25 files changed, 1100 insertions(+), 217 deletions(-) create mode 100644 backend/app/price_limits.py create mode 100644 backend/app/share_capital.py create mode 100644 backend/tests/test_financial_shares.py create mode 100644 backend/tests/test_price_limits.py diff --git a/backend/app/api/financials.py b/backend/app/api/financials.py index b69d00d..15579bd 100644 --- a/backend/app/api/financials.py +++ b/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 逐行解析,更简单可靠)。 """ diff --git a/backend/app/api/kline.py b/backend/app/api/kline.py index dbf58d3..b4cfe1a 100644 --- a/backend/app/api/kline.py +++ b/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") - diff --git a/backend/app/backtest/engine.py b/backend/app/backtest/engine.py index be839db..7a6cdb7 100644 --- a/backend/app/backtest/engine.py +++ b/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( diff --git a/backend/app/backtest/matrix.py b/backend/app/backtest/matrix.py index 6dbcac7..6398797 100644 --- a/backend/app/backtest/matrix.py +++ b/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], *, diff --git a/backend/app/data_providers/custom/provider.py b/backend/app/data_providers/custom/provider.py index 2266892..53b258e 100644 --- a/backend/app/data_providers/custom/provider.py +++ b/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, diff --git a/backend/app/indicators/pipeline.py b/backend/app/indicators/pipeline.py index f03b501..bad5a3a 100644 --- a/backend/app/indicators/pipeline.py +++ b/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) 回退自算理论价。 diff --git a/backend/app/price_limits.py b/backend/app/price_limits.py new file mode 100644 index 0000000..a1cc35a --- /dev/null +++ b/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 diff --git a/backend/app/services/financial_analyzer.py b/backend/app/services/financial_analyzer.py index 850f430..cd5c597 100644 --- a/backend/app/services/financial_analyzer.py +++ b/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")] diff --git a/backend/app/services/financial_sync.py b/backend/app/services/financial_sync.py index daa20b5..fe0b5a3 100644 --- a/backend/app/services/financial_sync.py +++ b/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 diff --git a/backend/app/services/quote_service.py b/backend/app/services/quote_service.py index 81ecb8c..c706f8a 100644 --- a/backend/app/services/quote_service.py +++ b/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(): diff --git a/backend/app/services/screener.py b/backend/app/services/screener.py index 608ca07..1e62e29 100644 --- a/backend/app/services/screener.py +++ b/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] diff --git a/backend/app/share_capital.py b/backend/app/share_capital.py new file mode 100644 index 0000000..e882ab4 --- /dev/null +++ b/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", + ) diff --git a/backend/app/strategy/builtin/near_limit_up.py b/backend/app/strategy/builtin/near_limit_up.py index 519d64d..467ab91 100644 --- a/backend/app/strategy/builtin/near_limit_up.py +++ b/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)) diff --git a/backend/app/tickflow/repository.py b/backend/app/tickflow/repository.py index 24eeba0..ab941bb 100644 --- a/backend/app/tickflow/repository.py +++ b/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 diff --git a/backend/tests/backtest/test_matrix_strategy.py b/backend/tests/backtest/test_matrix_strategy.py index ab92e82..fc7815f 100644 --- a/backend/tests/backtest/test_matrix_strategy.py +++ b/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(): diff --git a/backend/tests/test_financial_shares.py b/backend/tests/test_financial_shares.py new file mode 100644 index 0000000..a324f5b --- /dev/null +++ b/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) diff --git a/backend/tests/test_price_limits.py b/backend/tests/test_price_limits.py new file mode 100644 index 0000000..822ff51 --- /dev/null +++ b/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", + } diff --git a/backend/tests/test_st_limit_and_sharpe.py b/backend/tests/test_st_limit_and_sharpe.py index 78502c9..75402b5 100644 --- a/backend/tests/test_st_limit_and_sharpe.py +++ b/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}, diff --git a/frontend/src/components/EChartsIntraday.tsx b/frontend/src/components/EChartsIntraday.tsx index 812ef13..b4ba6a8 100644 --- a/frontend/src/components/EChartsIntraday.tsx +++ b/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(null) const chartRef = useRef(null) const roRef = useRef(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 () => { diff --git a/frontend/src/components/StockIntradayChart.tsx b/frontend/src/components/StockIntradayChart.tsx index 7055714..22afd75 100644 --- a/frontend/src/components/StockIntradayChart.tsx +++ b/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} /> )} diff --git a/frontend/src/components/financials/StockFinancialDetail.tsx b/frontend/src/components/financials/StockFinancialDetail.tsx index 18e8c12..a4bb66c 100644 --- a/frontend/src/components/financials/StockFinancialDetail.tsx +++ b/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 = { { 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) { {/* 标签页 */} -
+
{TABS.map(t => { const Icon = t.icon const isActive = tab === t.key diff --git a/frontend/src/lib/api.ts b/frontend/src/lib/api.ts index f0beee6..d1febb3 100644 --- a/frontend/src/lib/api.ts +++ b/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 } }>( diff --git a/frontend/src/lib/useFinancials.ts b/frontend/src/lib/useFinancials.ts index 5758a6b..5c11e20 100644 --- a/frontend/src/lib/useFinancials.ts +++ b/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({ diff --git a/frontend/src/pages/Financials.tsx b/frontend/src/pages/Financials.tsx index 70741d9..0bb3035 100644 --- a/frontend/src/pages/Financials.tsx +++ b/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 = { income: '利润表', balance_sheet: '资产负债表', cash_flow: '现金流量表', + shares: '股本表', } const TABLE_ICON: Record = { @@ -24,12 +25,13 @@ const TABLE_ICON: Record = { 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 ( <> - +
@@ -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() { <> @@ -165,7 +167,7 @@ export function Financials() { {isFullSync - ? `已同步 ${syncedCount}/4 张表…` + ? `已同步 ${syncedCount}/${TABLE_ORDER.length} 张表…` : isSingleSync ? `同步${TABLE_LABELS[syncSingleTable!] ?? syncSingleTable}…` : '同步中…'} @@ -186,18 +188,18 @@ export function Financials() { } /> -
+
{syncing && (
- 正在从 TickFlow 拉取财务数据，请稍候… + 正在从财务数据源拉取数据，请稍候…
)} {/* 同步状态卡片 —— 始终显示,反映本地财务数据概况 */} {!isLoading && available && (
-
+
{Object.entries(TABLE_LABELS).map(([key, label]) => { const info = tables[key] const TIcon = TABLE_ICON[key] ?? Database @@ -316,7 +318,7 @@ export function Financials() { )}
diff --git a/frontend/src/pages/Indices.tsx b/frontend/src/pages/Indices.tsx index 460bcb4..4496834 100644 --- a/frontend/src/pages/Indices.tsx +++ b/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}