191 lines
7.5 KiB
Python
191 lines
7.5 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""情绪温度计四指标单测(2026-09-03,消费 provider.get_limit_pool 出口)。
|
|
|
|
四条口径契约与数据 session 2026-09-03 对齐,逐条钉死:
|
|
① 炸板率 = zbgc行数/(zt行数+zbgc行数)——涨停池含「炸过又回封」的票
|
|
(break_count>0 仍在 zt),两池并集=当日触板全集,不得用 zt 内
|
|
break_count>0 把回封票从分母里扣掉;
|
|
② dtgc 空表 = 当日 0 跌停,是合法值非缺数;
|
|
③ (history)trading_days_only 语义由 provider 承担,本层只按 trade_date 对齐;
|
|
④ 连板高度 = zt.consecutive_boards 取 max;行业集中度用 zt.industry 现成列。
|
|
"""
|
|
from typing import Any, Dict, Optional
|
|
|
|
import pandas as pd
|
|
import pytest
|
|
|
|
from sanguo_portfolio.sentiment_thermometer import (
|
|
compute_thermometer,
|
|
daily_thermometer,
|
|
history_thermometer,
|
|
)
|
|
|
|
_ZT_COLS = ["code", "name", "consecutive_boards", "break_count", "industry"]
|
|
_ZBGC_COLS = ["code", "name", "break_count", "industry"]
|
|
_DTGC_COLS = ["code", "name", "industry"]
|
|
|
|
|
|
def _df(cols, rows):
|
|
return pd.DataFrame(rows, columns=cols)
|
|
|
|
|
|
def _zt_rows(n, boards=(1,), industries=("电子",), resealed_idx=()):
|
|
"""n 行 zt;resealed_idx 行 break_count>0(炸过又回封,当日仍算触板后封住)。"""
|
|
return [
|
|
(f"00000{i}.SZ", f"s{i}", boards[i % len(boards)],
|
|
2 if i in resealed_idx else 0, industries[i % len(industries)])
|
|
for i in range(n)
|
|
]
|
|
|
|
|
|
class _FakeProvider:
|
|
"""按 kind→DataFrame 回放的假 provider(get_limit_pool 出口形状)。"""
|
|
|
|
def __init__(self, panels: Dict[str, pd.DataFrame]):
|
|
self._panels = panels
|
|
self.calls: list = []
|
|
|
|
def get_limit_pool(self, kind, date=None, start=None, end=None,
|
|
trading_days_only=True):
|
|
self.calls.append((kind, date, start, end))
|
|
return self._panels[kind]
|
|
|
|
|
|
# ---------------- compute_thermometer:纯函数四契约 ----------------
|
|
|
|
def test_broken_rate_contract_resealed_tickets_stay_in_denominator():
|
|
"""契约①:回封票留在分母。zt=7(其中 1 票 break_count=2)+zbgc=3
|
|
→ 炸板率=3/10;错口径 3/9(扣回封票)必须不出现。"""
|
|
zt = _df(_ZT_COLS, _zt_rows(7, resealed_idx=(0,)))
|
|
zbgc = _df(_ZBGC_COLS, [
|
|
("100001.SZ", "b1", 1, "券商"), ("100002.SZ", "b2", 2, "电子"),
|
|
("100003.SZ", "b3", 1, "白酒")])
|
|
out = compute_thermometer(zt, zbgc, _df(_DTGC_COLS, []))
|
|
assert out["broken_board_rate"] == pytest.approx(3 / 10)
|
|
assert out["limit_up_count"] == 7
|
|
|
|
|
|
def test_dtgc_empty_means_zero_limit_down_not_missing():
|
|
"""契约②:dtgc 标准列空表 → 跌停数 0(合法值);净涨停=涨停数;比值为 None。"""
|
|
zt = _df(_ZT_COLS, _zt_rows(4))
|
|
out = compute_thermometer(zt, _df(_ZBGC_COLS, []), _df(_DTGC_COLS, []))
|
|
assert out["limit_down_count"] == 0
|
|
assert out["net_limit"] == 4
|
|
assert out["limit_up_down_ratio"] is None
|
|
|
|
|
|
def test_max_boards_and_industry_concentration():
|
|
"""契约④:连板高度=consecutive_boards 之 max;行业集中度=top1 占比。"""
|
|
zt = _df(_ZT_COLS, _zt_rows(
|
|
5, boards=(1, 3, 2, 3, 1), industries=("电子", "电子", "券商", "电子", "白酒")))
|
|
out = compute_thermometer(zt, _df(_ZBGC_COLS, []), _df(_DTGC_COLS, []))
|
|
assert out["max_consecutive_boards"] == 3
|
|
assert out["top_industry"] == "电子"
|
|
assert out["industry_top1_ratio"] == pytest.approx(3 / 5)
|
|
|
|
|
|
def test_max_boards_coerces_string_values():
|
|
"""akshare 原样落盘的连板数可能是字符串,pd.to_numeric 吸收。"""
|
|
zt = _df(_ZT_COLS, [
|
|
("0000001.SZ", "s1", "5", 0, "电子"),
|
|
("0000002.SZ", "s2", "2", 0, "电子")])
|
|
out = compute_thermometer(zt, _df(_ZBGC_COLS, []), _df(_DTGC_COLS, []))
|
|
assert out["max_consecutive_boards"] == 5
|
|
|
|
|
|
def test_all_empty_pools_is_cold_not_crash():
|
|
"""三池全空(极端冰点)=全零指标,不抛异常。"""
|
|
out = compute_thermometer(
|
|
_df(_ZT_COLS, []), _df(_ZBGC_COLS, []), _df(_DTGC_COLS, []))
|
|
assert out == {
|
|
"limit_up_count": 0, "limit_down_count": 0, "net_limit": 0,
|
|
"limit_up_down_ratio": None, "max_consecutive_boards": 0,
|
|
"broken_board_rate": 0.0, "top_industry": None,
|
|
"industry_top1_ratio": 0.0,
|
|
}
|
|
|
|
|
|
def test_industry_blank_values_excluded():
|
|
"""行业列空串/NaN 不进集中度分母。"""
|
|
zt = _df(_ZT_COLS, [
|
|
("0000001.SZ", "s1", 1, 0, "电子"),
|
|
("0000002.SZ", "s2", 1, 0, ""),
|
|
("0000003.SZ", "s3", 1, 0, None),
|
|
("0000004.SZ", "s4", 1, 0, "券商")])
|
|
out = compute_thermometer(zt, _df(_ZBGC_COLS, []), _df(_DTGC_COLS, []))
|
|
assert out["top_industry"] == "电子"
|
|
assert out["industry_top1_ratio"] == pytest.approx(1 / 2)
|
|
|
|
|
|
# ---------------- daily_thermometer:出口包装 ----------------
|
|
|
|
def test_daily_thermometer_passes_kind_and_date():
|
|
"""三 kind 单日拉取透传;date 显式给定时回填 trade_date。"""
|
|
prov = _FakeProvider({
|
|
"zt": _df(_ZT_COLS + ["trade_date"], [r + ("2026-09-02",) for r in _zt_rows(6)]),
|
|
"zbgc": _df(_ZBGC_COLS, []),
|
|
"dtgc": _df(_DTGC_COLS + ["trade_date"], []),
|
|
})
|
|
out = daily_thermometer(prov, date="2026-09-02")
|
|
assert [(k, d) for k, d, s, e in prov.calls] == [
|
|
("zt", "2026-09-02"), ("zbgc", "2026-09-02"), ("dtgc", "2026-09-02")]
|
|
assert out["trade_date"] == "2026-09-02"
|
|
assert out["limit_up_count"] == 6
|
|
|
|
|
|
def test_daily_thermometer_date_none_uses_latest_from_df():
|
|
"""date=None(最新落盘日语义)时 trade_date 取 zt 尾行。"""
|
|
prov = _FakeProvider({
|
|
"zt": _df(_ZT_COLS + ["trade_date"],
|
|
[r + ("2026-08-31",) for r in _zt_rows(3)]),
|
|
"zbgc": _df(_ZBGC_COLS, []),
|
|
"dtgc": _df(_DTGC_COLS, []),
|
|
})
|
|
out = daily_thermometer(prov)
|
|
assert out["trade_date"] == "2026-08-31"
|
|
|
|
|
|
# ---------------- history_thermometer:阈值校准用 ----------------
|
|
|
|
def _panel(kind, rows):
|
|
cols = {"zt": _ZT_COLS, "zbgc": _ZBGC_COLS, "dtgc": _DTGC_COLS}[kind]
|
|
return _df(cols + ["trade_date"], rows)
|
|
|
|
|
|
def test_history_aligns_by_trade_date():
|
|
"""契约③:区间三 kind 各拉一次,按 trade_date 对齐逐日出指标;
|
|
某日某池无行(如跌停 0)按 0 计,不丢该日。"""
|
|
prov = _FakeProvider({
|
|
"zt": pd.concat([
|
|
_panel("zt", [r + ("2026-09-01",) for r in _zt_rows(5, boards=(2,))]),
|
|
_panel("zt", [r + ("2026-09-02",) for r in _zt_rows(8, boards=(1, 4))]),
|
|
], ignore_index=True),
|
|
"zbgc": pd.concat([
|
|
_panel("zbgc", [("100001.SZ", "b1", 1, "电子", "2026-09-01")]),
|
|
_panel("zbgc", []),
|
|
], ignore_index=True),
|
|
"dtgc": pd.concat([
|
|
_panel("dtgc", []),
|
|
_panel("dtgc", [("200001.SZ", "d1", "医药", "2026-09-02")]),
|
|
], ignore_index=True),
|
|
})
|
|
hist = history_thermometer(prov, start="2026-09-01", end="2026-09-02")
|
|
assert len(hist) == 2
|
|
assert list(hist["trade_date"]) == ["2026-09-01", "2026-09-02"]
|
|
row1 = hist.iloc[0]
|
|
assert row1["limit_up_count"] == 5
|
|
assert row1["broken_board_rate"] == pytest.approx(1 / 6)
|
|
assert row1["limit_down_count"] == 0
|
|
row2 = hist.iloc[1]
|
|
assert row2["limit_up_count"] == 8
|
|
assert row2["max_consecutive_boards"] == 4
|
|
assert row2["limit_down_count"] == 1
|
|
assert row2["net_limit"] == 7
|
|
|
|
|
|
def test_history_all_empty_returns_empty_frame():
|
|
prov = _FakeProvider({"zt": _df(_ZT_COLS, []), "zbgc": _df(_ZBGC_COLS, []),
|
|
"dtgc": _df(_DTGC_COLS, [])})
|
|
hist = history_thermometer(prov, start="2026-09-01", end="2026-09-30")
|
|
assert hist.empty
|