mirror of
https://ghfast.top/https://github.com/aeroxw/tick-stock-panel.git
synced 2026-09-12 17:54:15 +08:00
fix(indicators): 量比改为同花顺/东财标准算法 + 行情时间戳落盘 (#91)
原 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)。
This commit is contained in:
wshy
@@ -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))
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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)
|
||||
|
||||
|
||||
@@ -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("增量计算结果为空, 回退到全量计算")
|
||||
|
||||
@@ -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"),
|
||||
|
||||
Reference in New Issue
Block a user