From 0eb3e54b673268bb80b44dec8a9c3c4e5d93f55a Mon Sep 17 00:00:00 2001 From: shy3130 <415333856@qq.com> Date: Mon, 24 Aug 2026 17:14:55 +0800 Subject: [PATCH] =?UTF-8?q?fix(data):=20=E8=87=AA=E5=AE=9A=E4=B9=89?= =?UTF-8?q?=E6=BA=90=E5=AE=9E=E6=97=B6=E8=A1=8C=E6=83=85=E7=99=BE=E5=88=86?= =?UTF-8?q?=E5=88=B6=E6=B6=A8=E8=B7=8C=E5=B9=85=E6=88=AA=E9=9D=A2=E8=87=AA?= =?UTF-8?q?=E5=8A=A8=E5=BD=92=E4=B8=80=E4=B8=BA=E5=B0=8F=E6=95=B0=E5=88=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 契约要求 change_pct/amplitude/turnover_rate 小数制(0.0366=3.66%), 但 a-stock-data 等第三方接口返回 3.66 表示 3.66%, 原样透传导致行业/ 概念统计与前端展示整体放大 100 倍(用户反馈)。get_realtime 摄取边界 按截面中位数判定(|值|中位数>0.31 必为百分制, 小数制受 30cm 涨跌停 约束不可能超过), 整批归一; 小样本退用最大值; 附 7 项回归测试与文档说明。 --- backend/app/data_providers/custom/provider.py | 27 +++++ backend/tests/test_custom_pct_units.py | 103 ++++++++++++++++++ docs/custom-data-source.md | 2 +- 3 files changed, 131 insertions(+), 1 deletion(-) create mode 100644 backend/tests/test_custom_pct_units.py diff --git a/backend/app/data_providers/custom/provider.py b/backend/app/data_providers/custom/provider.py index 6928895..0d6f031 100644 --- a/backend/app/data_providers/custom/provider.py +++ b/backend/app/data_providers/custom/provider.py @@ -34,6 +34,31 @@ _REQUIRED = { "financial": {"symbol"}, } +# 小数制下 change_pct/amplitude/turnover_rate 的物理上限: A股最大涨跌停 30% (+容差)。 +# 中位数口径下小数制批次不可能超过该值, 百分制批次(典型中位数 0.5~3)必然超过。 +_PCT_FRACTION_MAX = 0.31 + + +def _normalize_pct_units(df: pl.DataFrame) -> pl.DataFrame: + """百分制源自适应归一为小数制 (契约: change_pct/amplitude/turnover_rate 为小数, + 0.0366 = 3.66%)。不少第三方接口(如 a-stock-data)直接返回 3.66 表示 3.66%, + 若不归一, 下游(行业/概念统计、前端 x100 展示)会整体放大 100 倍。 + + 截面判定: 样本 >= 5 用 |值| 中位数(对个别无涨跌幅限制新股免疫), + 小样本退用最大值。整批同除 100, 避免逐值阈值在 0.3~1 区间的歧义。 + """ + for col in ("change_pct", "amplitude", "turnover_rate"): + if col not in df.columns: + continue + df = df.with_columns(pl.col(col).cast(pl.Float64, strict=False).alias(col)) + vals = df[col].drop_nulls().abs() + if vals.is_empty(): + continue + stat = vals.median() if vals.len() >= 5 else vals.max() + if stat > _PCT_FRACTION_MAX: + df = df.with_columns((pl.col(col) / 100).alias(col)) + return df + class GenericHTTPProvider: """HTTP-backed custom source. It only handles fetching and schema mapping.""" @@ -121,6 +146,8 @@ class GenericHTTPProvider: cfg = self._dataset("realtime") rows = self._request_rows(cfg) df = self._mapped_frame(cfg, rows) + # 百分制源(返回 3.66 表示 3.66%)截面归一为契约小数制 + df = _normalize_pct_units(df) if df.is_empty(): return [] return df.to_dicts() diff --git a/backend/tests/test_custom_pct_units.py b/backend/tests/test_custom_pct_units.py new file mode 100644 index 0000000..50aeab6 --- /dev/null +++ b/backend/tests/test_custom_pct_units.py @@ -0,0 +1,103 @@ +"""自定义源实时行情涨跌幅单位自适应归一测试。 + +契约要求 change_pct/amplitude/turnover_rate 用小数制 (0.0366 = 3.66%), +但不少第三方接口(如 a-stock-data)直接返回 3.66 表示 3.66%。未归一会把 +行业/概念统计与前端 x100 展示整体放大 100 倍(用户反馈)。 +""" +from __future__ import annotations + +import polars as pl +import pytest + +from app.data_providers.custom.config import CustomSourceConfig, DatasetConfig +from app.data_providers.custom.provider import GenericHTTPProvider, _normalize_pct_units + + +def _df(pcts, amps=None, turnovers=None): + data = {"change_pct": pcts} + if amps is not None: + data["amplitude"] = amps + if turnovers is not None: + data["turnover_rate"] = turnovers + return pl.DataFrame(data) + + +def test_percent_unit_batch_is_divided_by_100(): + out = _normalize_pct_units(_df( + [1.5, -2.2, 0.9, 2.8, -1.1, 0.6, 3.3, -0.8], + amps=[2.0, 3.5, 1.8, 4.0, 2.5, 1.2, 5.0, 1.6], + turnovers=[0.5, 1.2, 0.8, 2.0, 0.9, 0.4, 1.5, 0.7], + )) + assert out["change_pct"][0] == pytest.approx(0.015) + assert out["amplitude"][0] == pytest.approx(0.02) + assert out["turnover_rate"][0] == pytest.approx(0.005) + + +def test_fraction_unit_batch_untouched(): + pcts = [0.015, -0.022, 0.009, 0.028, -0.011, 0.006, 0.033, -0.008] + out = _normalize_pct_units(_df(pcts, amps=[0.02, 0.035, 0.018, 0.04, 0.025, 0.012, 0.05, 0.016])) + assert out["change_pct"].to_list() == pcts + assert out["amplitude"][0] == pytest.approx(0.02) + + +def test_limit_up_fraction_30cm_not_divided(): + # 北交所 30% 涨跌停的小数制极值不应被误判为百分制 + out = _normalize_pct_units(_df([0.30, 0.29, 0.28, 0.27, 0.26])) + assert out["change_pct"].to_list() == [0.30, 0.29, 0.28, 0.27, 0.26] + + +def test_small_batch_uses_max(): + # <5 样本退用最大值: 百分制小盘整批归一 + out = _normalize_pct_units(_df([0.5, 0.2])) + assert out["change_pct"].to_list() == [pytest.approx(0.005), pytest.approx(0.002)] + # 小数制小样本不动 + out2 = _normalize_pct_units(_df([0.005, 0.002])) + assert out2["change_pct"].to_list() == [0.005, 0.002] + + +def test_string_values_are_cast(): + out = _normalize_pct_units(_df(["1.5", "-2.2", "0.9", "2.8", "3.3", "0.6"])) + assert out["change_pct"][0] == pytest.approx(0.015) + + +def test_missing_or_null_columns_noop(): + out = _normalize_pct_units(pl.DataFrame({"close": [1.0, 2.0]})) + assert out.columns == ["close"] + out2 = _normalize_pct_units(_df([None, None, None, None, None, None])) + assert out2["change_pct"].null_count() == 6 + + +def _realtime_provider(rows): + provider = GenericHTTPProvider(CustomSourceConfig( + name="pct_source", + display_name="Pct Source", + datasets={"realtime": DatasetConfig( + url="https://example.test/realtime", + field_map={ + "code": "symbol", "price": "last_price", "pre_close": "prev_close", + "pct": "change_pct", "amp": "amplitude", "turnover": "turnover_rate", + }, + )}, + )) + provider._request_rows = lambda cfg, **kwargs: rows + return provider + + +def test_get_realtime_normalizes_percent_source(): + provider = _realtime_provider([ + {"code": "S1", "price": 10.0, "pre_close": 9.85, "pct": 1.52, "amp": 2.4, "turnover": 1.1}, + {"code": "S2", "price": 20.0, "pre_close": 20.44, "pct": -2.15, "amp": 3.1, "turnover": 0.8}, + {"code": "S3", "price": 30.0, "pre_close": 29.8, "pct": 0.67, "amp": 1.9, "turnover": 0.5}, + {"code": "S4", "price": 40.0, "pre_close": 38.9, "pct": 2.83, "amp": 4.2, "turnover": 2.0}, + {"code": "S5", "price": 50.0, "pre_close": 50.55, "pct": -1.09, "amp": 2.0, "turnover": 0.9}, + {"code": "S6", "price": 60.0, "pre_close": 59.64, "pct": 0.60, "amp": 1.6, "turnover": 0.7}, + ]) + try: + rows = provider.get_realtime() + finally: + provider.close() + by_sym = {r["symbol"]: r for r in rows} + assert by_sym["S1"]["change_pct"] == pytest.approx(0.0152) + assert by_sym["S1"]["amplitude"] == pytest.approx(0.024) + assert by_sym["S1"]["turnover_rate"] == pytest.approx(0.011) + assert by_sym["S2"]["change_pct"] == pytest.approx(-0.0215) diff --git a/docs/custom-data-source.md b/docs/custom-data-source.md index 0d949f5..e4cc73d 100644 --- a/docs/custom-data-source.md +++ b/docs/custom-data-source.md @@ -125,7 +125,7 @@ datasets: 建议实时接口额外提供 `amount`、`change_pct`、`change_amount`、`amplitude`、`turnover_rate`、`name`。缺失时部分字段会由 pipeline 回算,但精度取决于可用输入。 -`change_pct` 和 `amplitude` 使用小数制,例如 `0.0366` 表示 `3.66%`。 +`change_pct` 和 `amplitude` 使用小数制,例如 `0.0366` 表示 `3.66%`(`turnover_rate` 同)。若接口直接返回百分数值 `3.66`,实时行情会按截面中位数自动归一为小数制,但仍建议接口直接提供小数制以避免小样本歧义。 ## 请求约定