# -*- 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