diff --git a/sanguo_portfolio/sentiment_thermometer.py b/sanguo_portfolio/sentiment_thermometer.py
new file mode 100644
index 0000000..8539ca5
--- /dev/null
+++ b/sanguo_portfolio/sentiment_thermometer.py
@@ -0,0 +1,113 @@
+# -*- coding: utf-8 -*-
+"""情绪温度计——涨停池三件套四指标(消费 provider.get_limit_pool 出口)。
+
+原料:每日 19:30 ak-events 落盘的 zt/zbgc/dtgc 三池(get_limit_pool kind
+三合一,英文标准列 code/consecutive_boards/industry…,真空日=标准列空表)。
+
+四指标口径(与数据 session 契约对齐 2026-09-03):
+- 涨停/跌停数: 行数计;**dtgc 空=当日 0 跌停,是合法值非缺数**
+- 连板高度: zt.consecutive_boards 取 max(字符串值 to_numeric 吸收)
+- 炸板率: zbgc行数/(zt行数+zbgc行数)——涨停池含「炸过又回封」的票
+  (break_count>0 仍在 zt),两池并集=当日触板全集,勿用 zt 内
+  break_count>0 把回封票从分母里扣掉
+- 涨停行业集中度: zt.industry top1 占比+行业名(空串/NaN 剔除)
+
+定位:巡检/守卫层(veto/overlay)原料,不进信号层;阈值由 history_thermometer
+对已积累历史校准后另行确定,本模块只算不判。
+"""
+from typing import Any, Dict, Optional
+
+import pandas as pd
+
+
+def compute_thermometer(
+    zt: pd.DataFrame,
+    zbgc: pd.DataFrame,
+    dtgc: pd.DataFrame,
+) -> Dict[str, Any]:
+    """单日四指标(纯函数,输入 get_limit_pool 三 kind 的当日切版)。"""
+    zt_n, zbgc_n, dt_n = len(zt), len(zbgc), len(dtgc)
+    touch_total = zt_n + zbgc_n
+
+    max_boards = 0
+    if zt_n and "consecutive_boards" in zt.columns:
+        boards = pd.to_numeric(zt["consecutive_boards"], errors="coerce")
+        if boards.notna().any():
+            max_boards = int(boards.max())
+
+    if zt_n and "industry" in zt.columns:
+        ind = zt["industry"].dropna().astype(str).str.strip()
+        ind = ind[ind != ""]
+    else:
+        ind = pd.Series(dtype=object)
+    if len(ind):
+        counts = ind.value_counts()
+        top_industry: Optional[str] = str(counts.index[0])
+        top1_ratio: float = float(counts.iloc[0]) / float(len(ind))
+    else:
+        top_industry, top1_ratio = None, 0.0
+
+    return {
+        "limit_up_count": zt_n,
+        "limit_down_count": dt_n,
+        "net_limit": zt_n - dt_n,
+        "limit_up_down_ratio": (float(zt_n) / float(dt_n)) if dt_n else None,
+        "max_consecutive_boards": max_boards,
+        "broken_board_rate": (float(zbgc_n) / float(touch_total)) if touch_total else 0.0,
+        "top_industry": top_industry,
+        "industry_top1_ratio": round(top1_ratio, 4),
+    }
+
+
+def daily_thermometer(
+    provider: Any,
+    date: Optional[str] = None,
+) -> Dict[str, Any]:
+    """当日快照:三 kind 单日拉取(单日即 raw 语义,节假日副本不入场)。
+
+    date=None → provider 的最新落盘日语义,trade_date 从返回尾行解析。
+    """
+    zt = provider.get_limit_pool("zt", date=date)
+    zbgc = provider.get_limit_pool("zbgc", date=date)
+    dtgc = provider.get_limit_pool("dtgc", date=date)
+    result = compute_thermometer(zt, zbgc, dtgc)
+    if date is not None:
+        result["trade_date"] = str(date)
+    elif "trade_date" in zt.columns and len(zt):
+        result["trade_date"] = str(zt["trade_date"].iloc[-1])
+    else:
+        result["trade_date"] = None
+    return result
+
+
+def history_thermometer(
+    provider: Any,
+    start: str,
+    end: Optional[str] = None,
+) -> pd.DataFrame:
+    """区间逐日指标(阈值校准用):三 kind 各一次区间拉取,按 trade_date 对齐。
+
+    日期并集驱动——某池某日无行(如跌停 0)按 0 计,不丢该日;
+    trading_days_only 滤非交易日由 provider 承担,本层不再判历。
+    """
+    frames: Dict[str, pd.DataFrame] = {}
+    for kind in ("zt", "zbgc", "dtgc"):
+        df = provider.get_limit_pool(kind, start=start, end=end)
+        frames[kind] = df if isinstance(df, pd.DataFrame) else pd.DataFrame()
+
+    dates: set = set()
+    for df in frames.values():
+        if not df.empty and "trade_date" in df.columns:
+            dates.update(str(d) for d in df["trade_date"])
+
+    rows = []
+    for d in sorted(dates):
+        slices = {
+            kind: (df[df["trade_date"] == d] if not df.empty else df)
+            for kind, df in frames.items()
+        }
+        row = compute_thermometer(
+            slices["zt"], slices["zbgc"], slices["dtgc"])
+        row["trade_date"] = d
+        rows.append(row)
+    return pd.DataFrame(rows)
diff --git a/tests/portfolio/test_sentiment_thermometer.py b/tests/portfolio/test_sentiment_thermometer.py
new file mode 100644
index 0000000..cbdb8aa
--- /dev/null
+++ b/tests/portfolio/test_sentiment_thermometer.py
@@ -0,0 +1,190 @@
+# -*- 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