131 lines
4.7 KiB
Python
131 lines
4.7 KiB
Python
"""顶底「市场温度」透明启发式（可调，非黑盒）。
|
||
|
||
温度 0–100：越高越「过热/顶部风险」，越低越「冰点/底部机会」。
|
||
四个分量（缺哪个自动剔除并重新归一权重）：
|
||
vol_pct 量能分位：天量→高温(顶)，地量→低温(底)
|
||
price_pct 价格分位：高位→高温，低位→低温
|
||
sentiment 情绪：普涨/涨停潮亢奋→高温；普跌/跌停潮恐慌→低温
|
||
etf_heat ETF「出货热度」= (-净流入) 分位：大额净流出→高温(顶)，大额净流入(国家队)→低温(底)
|
||
|
||
权重与阈值来自 config（W_VOL/W_PRICE/W_SENTIMENT/W_ETF、HOT/COLD_THRESHOLD），
|
||
初值仅供起步，请用你自己的历史数据回测校准。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
from datetime import date
|
||
from typing import Dict, List, Optional
|
||
|
||
from .config import get_settings
|
||
|
||
settings = get_settings()
|
||
|
||
|
||
def _rolling_pct(values: List[Optional[float]], window: int) -> List[Optional[float]]:
|
||
"""滚动百分位排名(0-100)：当前值在最近 window 个有效值中的分位。"""
|
||
out: List[Optional[float]] = []
|
||
for i in range(len(values)):
|
||
cur = values[i]
|
||
if cur is None:
|
||
out.append(None)
|
||
continue
|
||
lo = max(0, i - window + 1)
|
||
base = [v for v in values[lo : i + 1] if v is not None]
|
||
if len(base) < max(5, window // 10): # 样本太少不给分位，避免早期噪声
|
||
out.append(None)
|
||
continue
|
||
le = sum(1 for v in base if v <= cur)
|
||
out.append(round(le / len(base) * 100, 2))
|
||
return out
|
||
|
||
|
||
def _norm_sentiment(sr) -> Optional[float]:
|
||
"""把一条情绪原始值折算到 0-100 的「亢奋度」粗分（缺字段则 None，交给分位再平滑）。
|
||
|
||
这里只做方向性折算：涨跌家数比越高、涨幅>5%家数越多 → 越亢奋。
|
||
真正的相对高低由外层 _rolling_pct 处理。
|
||
"""
|
||
if sr is None:
|
||
return None
|
||
parts = []
|
||
if sr.get("up_down_ratio") is not None:
|
||
parts.append(sr["up_down_ratio"])
|
||
if sr.get("pct_chg_gt_5_count") is not None:
|
||
parts.append(sr["pct_chg_gt_5_count"])
|
||
if not parts:
|
||
return None
|
||
return float(sum(parts)) # 量纲无所谓，后续走分位
|
||
|
||
|
||
def compute_signals(
|
||
bars: List[dict],
|
||
sentiment_by_date: Dict[date, dict],
|
||
etf_net_by_date: Dict[date, float],
|
||
window: Optional[int] = None,
|
||
) -> List[dict]:
|
||
"""bars: [{trade_date, close, amount}, ...] 升序。返回逐日温度与标记。"""
|
||
window = window or settings.vol_window
|
||
dates = [b["trade_date"] for b in bars]
|
||
closes = [b.get("close") for b in bars]
|
||
amounts = [b.get("amount") for b in bars]
|
||
|
||
# 情绪与 ETF 原始序列（对齐 bars 日期）
|
||
senti_raw = [_norm_sentiment(sentiment_by_date.get(d)) for d in dates]
|
||
etf_out_raw = [
|
||
(-etf_net_by_date[d]) if d in etf_net_by_date and etf_net_by_date[d] is not None else None
|
||
for d in dates
|
||
]
|
||
|
||
vol_pct = _rolling_pct(amounts, window)
|
||
price_pct = _rolling_pct(closes, window)
|
||
senti_pct = _rolling_pct(senti_raw, window)
|
||
etf_pct = _rolling_pct(etf_out_raw, window)
|
||
|
||
W = {
|
||
"vol": settings.w_vol,
|
||
"price": settings.w_price,
|
||
"sentiment": settings.w_sentiment,
|
||
"etf": settings.w_etf,
|
||
}
|
||
|
||
out: List[dict] = []
|
||
for i, d in enumerate(dates):
|
||
comps = {
|
||
"vol": vol_pct[i],
|
||
"price": price_pct[i],
|
||
"sentiment": senti_pct[i],
|
||
"etf": etf_pct[i],
|
||
}
|
||
avail = {k: v for k, v in comps.items() if v is not None}
|
||
if avail:
|
||
wsum = sum(W[k] for k in avail)
|
||
temp = round(sum(comps[k] * W[k] for k in avail) / wsum, 2) if wsum else None
|
||
else:
|
||
temp = None
|
||
|
||
flags = []
|
||
if vol_pct[i] is not None and price_pct[i] is not None:
|
||
if vol_pct[i] >= 95 and price_pct[i] >= 80:
|
||
flags.append("TOP_VOLUME_PRICE") # 天量见天价
|
||
if vol_pct[i] is not None and vol_pct[i] <= 5:
|
||
flags.append("BOTTOM_VOLUME_DRY") # 地量见地价
|
||
if senti_pct[i] is not None:
|
||
if senti_pct[i] >= 95:
|
||
flags.append("SENTIMENT_EUPHORIA")
|
||
elif senti_pct[i] <= 5:
|
||
flags.append("SENTIMENT_PANIC")
|
||
if temp is not None:
|
||
if temp >= settings.hot_threshold:
|
||
flags.append("OVERHEAT")
|
||
elif temp <= settings.cold_threshold:
|
||
flags.append("FREEZE")
|
||
|
||
out.append(
|
||
{
|
||
"trade_date": d,
|
||
"temperature": temp,
|
||
"components": comps,
|
||
"flags": flags,
|
||
}
|
||
)
|
||
return out
|