feat(strategy): 情绪温度计四指标——涨停池三件套首轮Python化,消费provider.get_limit_pool出口(get_limit_pool/get_event_panel双端8/8已验);四口径契约与数据session 09-03对齐并逐条钉进测试:①炸板率=zbgc/(zt+zbgc)按行数,回封票(zt内break_count>0)留分母勿扣②dtgc空=0跌停合法值③history按trade_date对齐,某池某日无行按0计不丢日④连板高度=consecutive_boards之max(to_numeric吸字符串)+行业集中度用industry现成列(空串/NaN剔除);三接口=compute纯函数+daily单日快照(date=None最新落盘日)+history区间校准表;只算不判,阈值待23日真实数据校准后另定;10新测+588全绿 [vps]
This commit is contained in:
@@ -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)
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user