From 1840e65f06a0864cb4789059bd3e30c4f0891310 Mon Sep 17 00:00:00 2001 From: wshy Date: Thu, 9 Jul 2026 16:41:02 +0800 Subject: [PATCH] =?UTF-8?q?fix(indicators):=20=E9=87=8F=E6=AF=94=E6=94=B9?= =?UTF-8?q?=E4=B8=BA=E5=90=8C=E8=8A=B1=E9=A1=BA/=E4=B8=9C=E8=B4=A2?= =?UTF-8?q?=E6=A0=87=E5=87=86=E7=AE=97=E6=B3=95=20+=20=E8=A1=8C=E6=83=85?= =?UTF-8?q?=E6=97=B6=E9=97=B4=E6=88=B3=E8=90=BD=E7=9B=98=20(#91)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 原 vol_ratio_5d 有两个叠加缺陷导致盘中严重偏低: 1. 分母含自身(rolling_mean(5) 算了当天) → 量比偏低约1.4倍 2. 盘中无时间折算(部分量直接比全天量级) → 开盘时段极低 改为标准量比公式(同花顺/东方财富/通达信一致, 已查证官方定义): 量比 = 今日累计成交量 / (前5日均量 × 已交易分钟数/240) - 分母不含当天(volume.shift(1).rolling_mean(5) / tail(5)前5日) - 盘中按已交易分钟数折算(×240/elapsed), 盘后系数=1.0不受影响 已交易分钟数用行情 quote_ts 推算(比服务端时间更准, 无网络延迟): - market_time.py: trading_minutes_elapsed_from_ts(毫秒时间戳) - normalizer.py: DAILY_COLS 加 quote_ts, 保留 SDK timestamp - quote_service.py: _build_daily 保留 quote_ts, 传入 elapsed_minutes - pipeline.py: 两路径(盘后全量/盘中增量)改为标准算法 + ENRICHED_STORAGE_COLS 加 quote_ts - repository.py: live_agg 新增 _vol_ma5_prev_sum(tail(5)前5日和) quote_ts 落盘后还可用于: 盘后收盘数据校验(非15:00重拉)、跨天完整性检查。 不改动: vol_ma5/vol_ma10保留原语义(含当天均量), 策略/选股阈值不变。 129后端测试全过, EOD量比验证正确(707636/100000=7.076)。 --- backend/app/data_providers/normalizer.py | 6 ++- backend/app/indicators/pipeline.py | 26 ++++++++-- backend/app/market_time.py | 60 +++++++++++++++++++++++- backend/app/services/quote_service.py | 16 ++++++- backend/app/tickflow/repository.py | 2 + 5 files changed, 102 insertions(+), 8 deletions(-) diff --git a/backend/app/data_providers/normalizer.py b/backend/app/data_providers/normalizer.py index edf047e..a43e716 100644 --- a/backend/app/data_providers/normalizer.py +++ b/backend/app/data_providers/normalizer.py @@ -5,7 +5,7 @@ import polars as pl from app.indicators.pipeline import filter_halt_days -DAILY_COLS = ["symbol", "date", "open", "high", "low", "close", "volume", "amount"] +DAILY_COLS = ["symbol", "date", "open", "high", "low", "close", "volume", "amount", "quote_ts"] ADJ_FACTOR_COLS = ["symbol", "trade_date", "ex_factor"] INSTRUMENT_COLS = ["symbol", "name", "code", "exchange", "asset_type", "source"] @@ -41,12 +41,16 @@ def normalize_daily(data, default_symbol: str | None = None, source: str = "tick "datetime": "date", "vol": "volume", "amt": "amount", + "timestamp": "quote_ts", } df = df.rename({k: v for k, v in rename_map.items() if k in df.columns}) if "symbol" not in df.columns and default_symbol: df = df.with_columns(pl.lit(default_symbol).alias("symbol")) if "date" in df.columns and df.schema["date"] != pl.Date: df = df.with_columns(pl.col("date").cast(pl.Date, strict=False)) + # quote_ts: 毫秒级行情时间戳, 用于盘后校验/量比折算。保留为 Int64, 缺失则置 null。 + if "quote_ts" in df.columns: + df = df.with_columns(pl.col("quote_ts").cast(pl.Int64, strict=False)) for col in ("open", "high", "low", "close", "volume", "amount"): if col in df.columns: df = df.with_columns(pl.col(col).cast(pl.Float64, strict=False)) diff --git a/backend/app/indicators/pipeline.py b/backend/app/indicators/pipeline.py index 5143842..3f7de8a 100644 --- a/backend/app/indicators/pipeline.py +++ b/backend/app/indicators/pipeline.py @@ -61,6 +61,7 @@ ENRICHED_STORAGE_COLS = [ "turnover_rate", # 依赖当时的 float_shares, 不可回推 "consecutive_limit_ups", # 递推状态, 需从历史 cum_sum "consecutive_limit_downs", + "quote_ts", # 行情时间戳(ms): 盘后校验/量比折算/跨天完整性 ] @@ -400,6 +401,9 @@ def compute_indicators(df: pl.DataFrame, needed: set[str] | None = None) -> pl.D _p1.append(pl.col("volume").rolling_mean(10).over("symbol").alias("vol_ma10")) if "_vol_ma5" in want: _p1.append(pl.col("volume").rolling_mean(5).over("symbol").alias("_vol_ma5")) + if "vol_ratio_5d" in want: + # 前5日平均成交量(不含当天), 标准量比分母: volume.shift(1).rolling_mean(5) + _p1.append(pl.col("volume").shift(1).rolling_mean(5).over("symbol").alias("_vol_ma5_prev")) if "high_60d" in want: _p1.append(pl.col("close").rolling_max(60).over("symbol").alias("high_60d")) if "low_60d" in want: @@ -450,8 +454,10 @@ def compute_indicators(df: pl.DataFrame, needed: set[str] | None = None) -> pl.D pl.col("_tr").ewm_mean(alpha=1.0 / 14, adjust=False).over("symbol").alias("atr_14"), ) if "vol_ratio_5d" in want: + # 标准量比(同花顺/东财): 今日成交量 / 前5日均量(不含当天) + # 盘后全量路径: 当日 volume 是完整全天量, 无需时间折算 df = df.with_columns( - (pl.col("volume") / pl.col("_vol_ma5")).alias("vol_ratio_5d"), + (pl.col("volume") / pl.col("_vol_ma5_prev")).alias("vol_ratio_5d"), ) _p4mom: list[pl.Expr] = [] if "momentum_5d" in want: @@ -516,7 +522,7 @@ def compute_indicators(df: pl.DataFrame, needed: set[str] | None = None) -> pl.D # 清理临时列 (只丢弃实际存在的临时列) _temp_cols = ["_boll_std", "_tr", "_ema12", "_ema26", - "_kdj_ln", "_kdj_hn", "_vol_ma5", "_daily_pct", + "_kdj_ln", "_kdj_hn", "_vol_ma5", "_vol_ma5_prev", "_daily_pct", "_delta", "_gain", "_loss", "_rsi_avg_gain_6", "_rsi_avg_loss_6", "_rsi_avg_gain_14", "_rsi_avg_loss_14", @@ -1226,6 +1232,7 @@ def compute_enriched_today( prev_enriched: pl.DataFrame, today_ohlcv: pl.DataFrame, instruments: pl.DataFrame | None = None, + elapsed_minutes: float | None = None, ) -> pl.DataFrame: """用昨天的递推状态 + 今天的 OHLCV 增量计算今天的 enriched 数据。 @@ -1236,6 +1243,8 @@ def compute_enriched_today( prev_enriched: repo.get_enriched_latest() — 昨天的完整 enriched (用于信号交叉判断) today_ohlcv: 今天的 OHLCV (symbol, date, open, high, low, close, volume, amount) instruments: 维表 (涨跌停/换手率需要) + elapsed_minutes: 当日已交易分钟数(用于标准量比的时间折算)。 + None 或 0 表示不折算(盘后或时间不可用, 此时 volume 已是全天量)。 返回: 今天的 enriched DataFrame (~5500 行, 64 列) @@ -1371,12 +1380,21 @@ def compute_enriched_today( ]) # ---- 量比 ---- + # vol_ma5/vol_ma10 保留原语义(含当天的均量), 其他地方在用 vol_ma5 = (pl.col("_vol_ma5_partial_sum") + pl.col("volume")) / 5 vol_ma10 = (pl.col("_vol_ma10_partial_sum") + pl.col("volume")) / 10 + # 标准量比(同花顺/东财): 今日累计成交量 / (前5日均量 × 已交易分钟数/240) + # _vol_ma5_prev_sum 是前5个交易日成交量之和(tail(5)), 不含当天 + # 盘中 volume 是部分量, 按 elapsed_minutes 折算到全天量级 + vol_ma5_prev = pl.col("_vol_ma5_prev_sum") / 5 # 前5日均量(不含当天) + if elapsed_minutes and elapsed_minutes > 0: + time_factor = 240.0 / elapsed_minutes # 盘中折算: 部分量 → 全天量级 + else: + time_factor = 1.0 # 盘后/无效时间: 不折算(此时 volume 已是全天量) df = df.with_columns([ vol_ma5.alias("vol_ma5"), vol_ma10.alias("vol_ma10"), - (pl.col("volume") / vol_ma5).alias("vol_ratio_5d"), + ((pl.col("volume") * time_factor) / vol_ma5_prev).alias("vol_ratio_5d"), ]) # ---- 极值 60 日 ---- @@ -1471,7 +1489,7 @@ def compute_enriched_today( "_high_59d", "_low_59d", "_close_5d_ago", "_close_10d_ago", "_close_20d_ago", "_close_30d_ago", "_close_60d_ago", - "_vol_ma5_partial_sum", "_vol_ma10_partial_sum", + "_vol_ma5_partial_sum", "_vol_ma10_partial_sum", "_vol_ma5_prev_sum", "_kdj_8d_low", "_kdj_8d_high", "_window_len", "_rsi_avg_gain_6", "_rsi_avg_loss_6", diff --git a/backend/app/market_time.py b/backend/app/market_time.py index 17895e2..f98835a 100644 --- a/backend/app/market_time.py +++ b/backend/app/market_time.py @@ -6,10 +6,17 @@ """ from __future__ import annotations -from datetime import date, datetime, timedelta, timezone +from datetime import date, datetime, time as dt_time, timedelta, timezone CN_TZ = timezone(timedelta(hours=8)) +# A 股交易时段 (北京时间): 上午 9:30-11:30 (120 分钟) + 下午 13:00-15:00 (120 分钟) = 240 分钟 +_TRADING_TOTAL_MINUTES = 240 +_MORNING_START = dt_time(9, 30) +_MORNING_END = dt_time(11, 30) +_AFTERNOON_START = dt_time(13, 0) +_AFTERNOON_END = dt_time(15, 0) + def cn_now() -> datetime: """当前北京时间 (带时区)。""" @@ -19,3 +26,54 @@ def cn_now() -> datetime: def cn_today() -> date: """当前北京日期。""" return datetime.now(CN_TZ).date() + + +def trading_minutes_elapsed_from_dt(dt: datetime) -> float: + """根据北京时间 datetime 计算当日已交易分钟数。 + + 交易时段: 9:30-11:30 (0~120) + 13:00-15:00 (120~240)。 + - 开盘前 = 0; 午休(11:30-13:00) = 120(保持上午累计); 收盘后 = 240。 + - 非交易日(周末) = 240 (视作全天, 避免量比被折算成 0)。 + """ + t = dt.time() + if t < _MORNING_START: + return 0.0 + if t < _MORNING_END: + return (dt.hour * 60 + dt.minute - 9 * 60 - 30) + dt.second / 60.0 + if t < _AFTERNOON_START: + return 120.0 # 午休, 保持上午累计 + if t < _AFTERNOON_END: + return 120.0 + (dt.hour * 60 + dt.minute - 13 * 60) + dt.second / 60.0 + return float(_TRADING_TOTAL_MINUTES) + + +def trading_minutes_elapsed() -> float: + """当前已交易分钟数 (基于服务端北京时间)。 + + 量比折算的兜底: 当行情 timestamp 缺失时用服务端时间。 + 优先使用 trading_minutes_elapsed_from_ts (行情真实时间, 更准)。 + """ + return trading_minutes_elapsed_from_dt(cn_now()) + + +def trading_minutes_elapsed_from_ts(ts_ms: int | float | None) -> float: + """从行情时间戳(毫秒)计算当日已交易分钟数。 + + 优先使用此函数: 行情 timestamp 是真实成交时间, 比服务端时间更准 + (服务端时间含网络/限流延迟)。 + + Args: + ts_ms: 毫秒级 Unix 时间戳 (TickFlow SDK quote.timestamp / kline.timestamp) + + Returns: + 已交易分钟数 (0~240)。timestamp 为 None/无效时返回 240 (视作全天, + 避免量比被折算成 0)。 + """ + if not ts_ms: + return float(_TRADING_TOTAL_MINUTES) + try: + dt = datetime.fromtimestamp(int(ts_ms) / 1000.0, tz=CN_TZ) + except (ValueError, TypeError, OSError): + return float(_TRADING_TOTAL_MINUTES) + return trading_minutes_elapsed_from_dt(dt) + diff --git a/backend/app/services/quote_service.py b/backend/app/services/quote_service.py index b62ed0a..2457a5c 100644 --- a/backend/app/services/quote_service.py +++ b/backend/app/services/quote_service.py @@ -766,7 +766,7 @@ class QuoteService: @staticmethod def _build_daily(records: list[dict]) -> pl.DataFrame: - """将 API records 转为日K格式 DataFrame (只有 OHLCV, 写 kline_daily 用)。""" + """将 API records 转为日K格式 DataFrame (OHLCV + quote_ts, 写 kline_daily 用)。""" if not records: return pl.DataFrame() df = pl.DataFrame(records) @@ -778,11 +778,13 @@ class QuoteService: "low": "low", "volume": "volume", "amount": "amount", + "timestamp": "quote_ts", } select_exprs = [] for src, dst in cols_map.items(): if src in df.columns: - select_exprs.append(pl.col(src).alias(dst)) + select_exprs.append(pl.col(src).cast(pl.Int64, strict=False).alias(dst) + if dst == "quote_ts" else pl.col(src).alias(dst)) if not select_exprs: return pl.DataFrame() result = df.select(select_exprs).with_columns( @@ -1193,16 +1195,26 @@ class QuoteService: if use_incremental: from app.indicators.pipeline import compute_enriched_today + from app.market_time import trading_minutes_elapsed_from_ts, trading_minutes_elapsed instruments = self._repo.get_instruments() # 将 API 直接提供的补充字段 JOIN 到 daily_df today_ohlcv = daily_df if quote_extra is not None and not quote_extra.is_empty(): today_ohlcv = daily_df.join(quote_extra, on="symbol", how="left") + # 量比时间折算: 优先用行情 quote_ts (真实成交时间), 缺失则兜底服务端时间 + elapsed_minutes: float | None = None + if "quote_ts" in daily_df.columns and not daily_df.is_empty(): + valid_ts = daily_df["quote_ts"].drop_nulls() + if not valid_ts.is_empty(): + elapsed_minutes = trading_minutes_elapsed_from_ts(valid_ts.median()) + if elapsed_minutes is None: + elapsed_minutes = trading_minutes_elapsed() enriched_today = compute_enriched_today( live_agg=live_agg, prev_enriched=prev_enriched, today_ohlcv=today_ohlcv, instruments=instruments, + elapsed_minutes=elapsed_minutes, ) if enriched_today.is_empty(): logger.warning("增量计算结果为空, 回退到全量计算") diff --git a/backend/app/tickflow/repository.py b/backend/app/tickflow/repository.py index f54ffbe..f022ff6 100644 --- a/backend/app/tickflow/repository.py +++ b/backend/app/tickflow/repository.py @@ -748,6 +748,8 @@ class KlineRepository: pl.col("volume").tail(4).sum().alias("_vol_ma5_partial_sum"), pl.col("volume").tail(9).sum().alias("_vol_ma10_partial_sum"), + # 标准量比分母: 前5日成交量之和(不含当天), 用于 vol_ratio_5d + pl.col("volume").tail(5).sum().alias("_vol_ma5_prev_sum"), pl.col("low").tail(8).min().alias("_kdj_8d_low"), pl.col("high").tail(8).max().alias("_kdj_8d_high"),