aeroxw/tick-stock-panel
1
0
Fork 0
mirror of https://ghfast.top/https://github.com/aeroxw/tick-stock-panel.git synced 2026-09-12 16:44:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(data): 自定义源比例字段单位改为显式声明 pct_unit 未声明 fail-closed

Browse Source 
amplitude/turnover_rate 的百分制与小数制数值区间重叠(0.05 既可能是
0.05% 也可能是 5%), 截面中位数启发式不可判, 赌错即整体放大 100 倍,
违反 CONTRIBUTING §3.1 禁止启发式转换的约束。

- realtime 数据集新增 pct_unit: percent|decimal 显式声明, 声明即契约
  (percent 无条件 /100, decimal 无条件透传, 不受数值外观影响)
- 未声明时 change_pct 保留涨跌停 30% 上限的截面判定(物理可判),
  amplitude/turnover_rate 置 None 交 enriched 管道按价格/股本口径重算
  并记录 WARNING; 已配置 transforms 的列视为用户接管单位, 透传
- 配置解析/清洗/序列化全链路校验取值, 非 realtime 数据集声明即报错
- 契约测试重写覆盖声明优先、边界值、fail-closed 与 transforms 兼容
This commit is contained in:
shy3130
2026-08-30 19:05:20 +08:00
parent 0b4bde6dda
commit 578a531743
5 changed files with 330 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/data_providers/custom/config.py
+8
View File
@@ -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,
)
backend/app/data_providers/custom/loader.py
+8
View File
@@ -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()
backend/app/data_providers/custom/provider.py
+52 -15
View File
@@ -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()
backend/tests/test_custom_pct_units.py
+241 -33
View File
@@ -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)
docs/custom-data-source.md
+21 -3
View File
@@ -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 = 股票代码