diff --git a/backend/app/data_providers/custom/config.py b/backend/app/data_providers/custom/config.py index cf4b732..e5004b3 100644 --- a/backend/app/data_providers/custom/config.py +++ b/backend/app/data_providers/custom/config.py @@ -37,6 +37,9 @@ class DatasetConfig: end_param: str = "end_time" asset_type_param: str | None = None freq_param: str | None = None + # realtime 比例字段(change_pct/amplitude/turnover_rate)的单位声明: + # "percent"(返回 3.66 表示 3.66%)或 "decimal"(返回 0.0366 表示 3.66%)。 + pct_unit: str | None = None @dataclass(frozen=True) @@ -75,6 +78,10 @@ def _dataset_from_dict(raw: dict[str, Any]) -> DatasetConfig: if not 0 < timeout <= MAX_TIMEOUT: raise ValueError(f"timeout must be between 0 and {MAX_TIMEOUT:g} seconds") + pct_unit = str(raw.get("pct_unit") or "").strip().lower() or None + if pct_unit not in (None, "percent", "decimal"): + raise ValueError(f"pct_unit must be 'percent' or 'decimal', got {pct_unit!r}") + return DatasetConfig( url=str(raw.get("url", "") or ""), method=str(raw.get("method", "GET") or "GET").upper(), @@ -91,6 +98,7 @@ def _dataset_from_dict(raw: dict[str, Any]) -> DatasetConfig: end_param=str(raw.get("end_param", "end_time") or "end_time").strip() or "end_time", asset_type_param=(str(raw.get("asset_type_param") or "").strip() or None), freq_param=(str(raw.get("freq_param") or "").strip() or None), + pct_unit=pct_unit, ) diff --git a/backend/app/data_providers/custom/loader.py b/backend/app/data_providers/custom/loader.py index 4f3aa1d..2a5985b 100644 --- a/backend/app/data_providers/custom/loader.py +++ b/backend/app/data_providers/custom/loader.py @@ -353,6 +353,7 @@ def _config_to_dict(config: CustomSourceConfig) -> dict: } if ds_name != "realtime" else {}), **({"asset_type_param": ds.asset_type_param} if ds_name == "minute" and ds.asset_type_param else {}), **({"freq_param": ds.freq_param} if ds_name == "minute" and ds.freq_param else {}), + **({"pct_unit": ds.pct_unit} if ds_name == "realtime" and ds.pct_unit else {}), } return out @@ -476,6 +477,13 @@ def _sanitize_dataset(ds_name: str, ds_cfg: dict) -> dict: out["start_param"] = start_param if end_param: out["end_param"] = end_param + pct_unit = str(ds_cfg.get("pct_unit") or "").strip().lower() + if pct_unit: + if ds_name != "realtime": + raise ValueError(f"{ds_name}: pct_unit 仅用于 realtime 数据集") + if pct_unit not in ("percent", "decimal"): + raise ValueError(f"{ds_name}: pct_unit 必须是 percent 或 decimal") + out["pct_unit"] = pct_unit if ds_name == "minute": asset_type_param = str(ds_cfg.get("asset_type_param") or "").strip() freq_param = str(ds_cfg.get("freq_param") or "").strip() diff --git a/backend/app/data_providers/custom/provider.py b/backend/app/data_providers/custom/provider.py index 0d6f031..346e894 100644 --- a/backend/app/data_providers/custom/provider.py +++ b/backend/app/data_providers/custom/provider.py @@ -34,29 +34,56 @@ _REQUIRED = { "financial": {"symbol"}, } -# 小数制下 change_pct/amplitude/turnover_rate 的物理上限: A股最大涨跌停 30% (+容差)。 +# 小数制下 change_pct 的物理上限: A股最大涨跌停 30% (+容差)。 # 中位数口径下小数制批次不可能超过该值, 百分制批次(典型中位数 0.5~3)必然超过。 +# 仅对 change_pct 有效——amplitude/turnover_rate 的两种单位在数值区间上重叠 +# (百分制 0.05 = 0.05% 与小数制 0.05 = 5%), 无物理依据可判。 _PCT_FRACTION_MAX = 0.31 +_PCT_COLUMNS = ("change_pct", "amplitude", "turnover_rate") -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 区间的歧义。 +def _normalize_pct_units( + df: pl.DataFrame, + pct_unit: str | None = None, + transformed_cols: frozenset[str] = frozenset(), +) -> pl.DataFrame: + """比例字段单位归一为契约小数制 (change_pct/amplitude/turnover_rate, + 0.0366 = 3.66%, CONTRIBUTING §3.1)。单位只认显式声明, 不靠数值猜: + + - pct_unit="percent" → 三列无条件 /100 (声明即契约, 即使数值看着像小数制); + - pct_unit="decimal" → 原样透传 (即使数值看着像百分制也不动); + - 未声明 → change_pct 保留截面中位数判定(涨跌停 30% 上限使其物理可判: + 样本 >= 5 用 |值| 中位数, 小样本退用最大值, 整批同除 100); + amplitude/turnover_rate 置 None 交下游重算(enriched 管道按 + high/low/prev_close 与股本口径重算), 除非该列已被 transforms 显式 + 处理过(视为用户已接管单位, 原样透传)。 """ - for col in ("change_pct", "amplitude", "turnover_rate"): + dropped_undeclared = False + for col in _PCT_COLUMNS: 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: + if pct_unit == "percent": df = df.with_columns((pl.col(col) / 100).alias(col)) + elif pct_unit == "decimal" or col in transformed_cols: + continue + elif col == "change_pct": + 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)) + else: + df = df.with_columns(pl.lit(None, dtype=pl.Float64).alias(col)) + dropped_undeclared = True + if dropped_undeclared: + logger.warning( + "自定义源 realtime 未声明 pct_unit: amplitude/turnover_rate 的单位" + "无法从数值判定, 已置 None 交由下游按股本/价格口径重算;" + "请在 realtime 数据集配置中显式声明 pct_unit: percent 或 decimal" + ) return df @@ -82,6 +109,11 @@ class GenericHTTPProvider: missing = sorted(required - mapped) if missing: errors.append(f"{dataset}: missing mapped fields: {', '.join(missing)}") + if cfg.pct_unit is not None: + if dataset != "realtime": + errors.append(f"{dataset}: pct_unit 仅用于 realtime 数据集") + elif cfg.pct_unit not in ("percent", "decimal"): + errors.append(f"{dataset}: pct_unit 必须是 percent 或 decimal") if dataset != "realtime": request_params = [cfg.symbols_param, cfg.start_param, cfg.end_param] if dataset == "minute": @@ -146,8 +178,13 @@ 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) + # 单位归一: 显式 pct_unit 声明优先; 未声明时 amplitude/turnover_rate + # fail-closed 置 None(交下游重算), change_pct 保留截面判定 + df = _normalize_pct_units( + df, + pct_unit=cfg.pct_unit, + transformed_cols=frozenset(cfg.transforms) & set(_PCT_COLUMNS), + ) 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 index 50aeab6..616351d 100644 --- a/backend/tests/test_custom_pct_units.py +++ b/backend/tests/test_custom_pct_units.py @@ -1,15 +1,19 @@ -"""自定义源实时行情涨跌幅单位自适应归一测试。 +"""自定义源实时行情比例字段单位归一测试 (CONTRIBUTING §3.1)。 -契约要求 change_pct/amplitude/turnover_rate 用小数制 (0.0366 = 3.66%), -但不少第三方接口(如 a-stock-data)直接返回 3.66 表示 3.66%。未归一会把 -行业/概念统计与前端 x100 展示整体放大 100 倍(用户反馈)。 +契约: change_pct/amplitude/turnover_rate 为小数制 (0.0366 = 3.66%)。 +单位只认显式声明 pct_unit: percent|decimal, 不靠数值猜: + - 声明 percent → 无条件 /100; 声明 decimal → 无条件透传; + - 未声明 → change_pct 保留截面中位数判定(涨跌停 30% 上限物理可判), + amplitude/turnover_rate 置 None 交下游重算(fail-closed), + 已被 transforms 显式处理过的列视为用户接管单位, 透传。 """ + from __future__ import annotations import polars as pl import pytest -from app.data_providers.custom.config import CustomSourceConfig, DatasetConfig +from app.data_providers.custom.config import CustomSourceConfig, DatasetConfig, config_from_dict from app.data_providers.custom.provider import GenericHTTPProvider, _normalize_pct_units @@ -22,24 +26,64 @@ def _df(pcts, amps=None, turnovers=None): 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], - )) +# ---- 显式声明: percent ---- + + +def test_declared_percent_divides_all_columns(): + 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], + ), + pct_unit="percent", + ) 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(): +def test_declared_percent_wins_even_when_values_look_decimal(): + # 百分制低波动日: 0.25 表示 0.25%, 数值落在小数制区间内——声明优先, 不靠猜 + out = _normalize_pct_units( + _df( + [0.25, 0.30, 0.28, 0.27, 0.26, 0.22], + amps=[0.4, 0.5, 0.45, 0.6, 0.5, 0.4], + turnovers=[0.05, 0.08, 0.06, 0.1, 0.07, 0.05], + ), + pct_unit="percent", + ) + assert out["change_pct"][0] == pytest.approx(0.0025) + assert out["amplitude"][0] == pytest.approx(0.004) + assert out["turnover_rate"][0] == pytest.approx(0.0005) + + +# ---- 显式声明: decimal ---- + + +def test_declared_decimal_passes_through(): 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])) + out = _normalize_pct_units( + _df(pcts, amps=[0.02, 0.035, 0.018, 0.04, 0.025, 0.012, 0.05, 0.016]), pct_unit="decimal" + ) assert out["change_pct"].to_list() == pcts assert out["amplitude"][0] == pytest.approx(0.02) +def test_declared_decimal_wins_even_when_values_look_percent(): + # 用户声明了小数制就按小数制契约透传, 不替用户"修正"数据 + out = _normalize_pct_units(_df([3.66, -2.15, 0.9, 2.8, 1.1]), pct_unit="decimal") + assert out["change_pct"][0] == pytest.approx(3.66) + + +# ---- 未声明: change_pct 保留截面判定(物理可判) ---- + + +def test_undeclared_change_pct_percent_batch_normalized(): + 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_limit_up_fraction_30cm_not_divided(): # 北交所 30% 涨跌停的小数制极值不应被误判为百分制 out = _normalize_pct_units(_df([0.30, 0.29, 0.28, 0.27, 0.26])) @@ -60,38 +104,85 @@ def test_string_values_are_cast(): assert out["change_pct"][0] == pytest.approx(0.015) +# ---- 未声明: amplitude/turnover_rate fail-closed (核心修复) ---- + + +def test_undeclared_amplitude_and_turnover_are_nulled(): + # 百分制 0.05 = 0.05% 与小数制 0.05 = 5% 数值相同, 不可判定 → 置 None + out = _normalize_pct_units( + _df( + [1.5, -2.2, 0.9, 2.8, 3.3, 0.6], + amps=[2.0, 3.5, 1.8, 4.0, 5.0, 1.6], + turnovers=[0.05, 1.2, 0.8, 2.0, 1.5, 0.7], + ) + ) + assert out["amplitude"].null_count() == 6 + assert out["turnover_rate"].null_count() == 6 + # change_pct 仍正常归一 + assert out["change_pct"][0] == pytest.approx(0.015) + + +def test_undeclared_transformed_column_passes_through(): + # 用户已用 transforms 显式处理过单位(如 value / 100)的列: 视为接管, 不置 None + out = _normalize_pct_units( + _df([1.5, -2.2, 0.9, 2.8, 3.3, 0.6], turnovers=[0.005, 0.012, 0.008, 0.02, 0.015, 0.007]), + transformed_cols=frozenset({"turnover_rate"}), + ) + assert out["turnover_rate"][0] == pytest.approx(0.005) + # 未 transform 的 amplitude 仍 fail-closed + assert "amplitude" not in out.columns + + 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 + # 全 null 的不可判定列保持 null + out3 = _normalize_pct_units(_df([1.5, -2.2, 0.9, 2.8, 3.3, 0.6], turnovers=[None] * 6)) + assert out3["turnover_rate"].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 集成 ---- + + +def _realtime_provider(rows, **ds_kwargs): + 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", + }, + **ds_kwargs, + ) }, - )}, - )) + ) + ) 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}, - ]) +_ROWS = [ + {"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}, +] + + +def test_get_realtime_declared_percent_source(): + provider = _realtime_provider(_ROWS, pct_unit="percent") try: rows = provider.get_realtime() finally: @@ -101,3 +192,120 @@ def test_get_realtime_normalizes_percent_source(): 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) + + +def test_get_realtime_undeclared_nulls_ambiguous_columns(): + provider = _realtime_provider(_ROWS) + try: + rows = provider.get_realtime() + finally: + provider.close() + by_sym = {r["symbol"]: r for r in rows} + # change_pct 截面判定仍归一 + assert by_sym["S1"]["change_pct"] == pytest.approx(0.0152) + # 不可判定列 fail-closed + assert by_sym["S1"]["amplitude"] is None + assert by_sym["S1"]["turnover_rate"] is None + + +def test_get_realtime_transformed_turnover_kept(): + provider = _realtime_provider(_ROWS, transforms={"turnover_rate": "value / 100"}) + try: + rows = provider.get_realtime() + finally: + provider.close() + by_sym = {r["symbol"]: r for r in rows} + assert by_sym["S1"]["turnover_rate"] == pytest.approx(0.011) + assert by_sym["S1"]["amplitude"] is None + + +# ---- 配置解析与校验 ---- + + +def test_config_parses_pct_unit(): + cfg = config_from_dict( + { + "name": "s", + "datasets": { + "realtime": { + "url": "https://example.test", + "pct_unit": "Percent", + } + }, + } + ) + assert cfg.datasets["realtime"].pct_unit == "percent" + + +def test_config_rejects_invalid_pct_unit(): + with pytest.raises(ValueError, match="pct_unit"): + config_from_dict( + { + "name": "s", + "datasets": { + "realtime": { + "url": "https://example.test", + "pct_unit": "basis_point", + } + }, + } + ) + + +def test_validate_flags_pct_unit_on_non_realtime(): + provider = GenericHTTPProvider( + CustomSourceConfig( + name="s", + display_name="S", + datasets={ + "daily": DatasetConfig( + url="https://example.test", + field_map={ + "c": "symbol", + "d": "date", + "o": "open", + "h": "high", + "l": "low", + "cl": "close", + "v": "volume", + "a": "amount", + }, + pct_unit="percent", + ) + }, + ) + ) + try: + errors = provider.validate() + finally: + provider.close() + assert any("pct_unit" in e and "realtime" in e for e in errors) + + +def test_validate_flags_invalid_pct_unit_value(): + provider = GenericHTTPProvider( + CustomSourceConfig( + name="s", + display_name="S", + datasets={ + "realtime": DatasetConfig( + url="https://example.test", + field_map={ + "c": "symbol", + "p": "last_price", + "pc": "prev_close", + "o": "open", + "h": "high", + "l": "low", + "v": "volume", + }, + pct_unit="bp", + ) + }, + ) + ) + try: + errors = provider.validate() + finally: + provider.close() + assert any("pct_unit" in e for e in errors) diff --git a/docs/custom-data-source.md b/docs/custom-data-source.md index e4cc73d..ae2f1b1 100644 --- a/docs/custom-data-source.md +++ b/docs/custom-data-source.md @@ -125,7 +125,23 @@ datasets: 建议实时接口额外提供 `amount`、`change_pct`、`change_amount`、`amplitude`、`turnover_rate`、`name`。缺失时部分字段会由 pipeline 回算,但精度取决于可用输入。 -`change_pct` 和 `amplitude` 使用小数制,例如 `0.0366` 表示 `3.66%`(`turnover_rate` 同)。若接口直接返回百分数值 `3.66`,实时行情会按截面中位数自动归一为小数制,但仍建议接口直接提供小数制以避免小样本歧义。 +`change_pct`、`amplitude`、`turnover_rate` 统一使用小数制,例如 `0.0366` 表示 `3.66%`。百分制单位必须在 realtime 数据集上**显式声明**,不做数值猜测(数值无法区分两种单位:`0.05` 既可能是 0.05% 也可能是 5%): + +```yaml +datasets: + realtime: + url: https://api.example.com/snapshot + pct_unit: percent # 接口返回 3.66 表示 3.66%;小数制源声明 decimal 或省略 +``` + +处理规则: + +| 声明 | 行为 | +| --- | --- | +| `pct_unit: percent` | `change_pct` / `amplitude` / `turnover_rate` 无条件 `/100` | +| `pct_unit: decimal` | 三列原样透传 | +| 未声明 | `change_pct` 按截面中位数归一(A 股涨跌停 30% 上限使两种单位物理可分);`amplitude` / `turnover_rate` **置 `None`** 交由 pipeline 按价格与股本口径重算,并记录 WARNING | +| 列已配置 `transforms` | 视为用户已接管该列单位,原样透传 | ## 请求约定 @@ -273,8 +289,10 @@ cp docs/examples/custom-data-source/mock_source.yaml data/data_sources/mock_sour amount = 成交额 change_pct = 涨跌幅 (小数, 0.0366 = 3.66%) change_amount = 涨跌额 - amplitude = 振幅 - turnover_rate = 换手率 (小数, 0.05 = 5%; 若上游返回 5 表示 5%, 配置 transforms: turnover_rate: "value / 100") + amplitude = 振幅 (小数, 0.024 = 2.4%) + turnover_rate = 换手率 (小数, 0.05 = 5%) + # 上游若返回百分数值 (3.66 表示 3.66%), 在 realtime 数据集声明 pct_unit: percent, + # 不要依赖数值自动识别; 逐列转换也可用 transforms: turnover_rate: "value / 100" 分钟K (minute): symbol = 股票代码