128 lines
5.2 KiB
Python
128 lines
5.2 KiB
Python
"""Stable signal taxonomy for recommendation analytics.
|
|
|
|
Display labels are allowed to change often; these codes are the analytics keys
|
|
used by review, signal performance, and candidate-rule generation.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from typing import Any, Iterable
|
|
|
|
|
|
SIGNAL_CODE_LABELS = {
|
|
"vp_fly_1h_current": "1H当前量价齐飞",
|
|
"volume_consecutive_1h": "1H连续放量",
|
|
"cex_top_gainer_24h": "CEX 24h强势榜异动",
|
|
"vp_fly_1h_stale": "1H历史量价齐飞",
|
|
"short_tf_15m_ignition": "15min短周期启动",
|
|
"short_tf_5m_ignition": "5min极早期启动",
|
|
"short_tf_resonance": "短周期共振",
|
|
"volume_divergence_1h": "1H量价背离",
|
|
"static_accum_4h": "4H静K蓄力",
|
|
"higher_lows_4h": "4H底部抬高",
|
|
"compression_surge_4h": "4H压缩放量",
|
|
"box_breakout_pullback_1h": "1H箱体突破回踩",
|
|
"box_breakout_pullback_4h": "4H箱体突破回踩",
|
|
"ignition_1h_current": "1H当前起爆点",
|
|
"ignition_4h_current": "4H当前起爆点",
|
|
"ignition_d1_current": "日线当前起爆点",
|
|
"ignition_stale": "历史起爆点",
|
|
"dynamic_k_1h_bull": "1H多头动K",
|
|
"dynamic_k_d1_bull": "日线多头动K",
|
|
"breakout_pullback_d1": "日线突破回踩",
|
|
"breakout_pullback_w1": "周线突破回踩",
|
|
"breakout_15m_current": "15min当前突破",
|
|
"pullback_15m_confirm": "15min回踩确认",
|
|
"strong_resonance_bypass": "强共振旁路",
|
|
"entry_quality_gate": "买点质量闸门",
|
|
"top_trader_long": "大户偏多",
|
|
"sector_rotation": "板块联动",
|
|
"sentiment_resonance": "舆情共振",
|
|
"dex_volume_spike": "DEX 放量",
|
|
"liquidity_add": "流动性增加",
|
|
"liquidity_remove_risk": "流动性撤出风险",
|
|
"exchange_outflow": "交易所流出",
|
|
"exchange_inflow_risk": "交易所流入风险",
|
|
"whale_accumulation": "鲸鱼增持",
|
|
"holder_concentration_risk": "持仓集中风险",
|
|
"smart_money_buying": "聪明钱买入",
|
|
"funding_extreme": "资金费率极端",
|
|
"trend_exhaustion": "趋势衰减",
|
|
"false_breakout": "假突破",
|
|
"high_position_reject": "高位拒绝",
|
|
"risk_reward_bad": "盈亏比不合格",
|
|
"unknown": "未分类信号",
|
|
}
|
|
|
|
|
|
_PATTERNS = [
|
|
("cex_top_gainer_24h", ("24h强势榜",)),
|
|
("vp_fly_1h_stale", ("历史放量阳线", "历史量价齐飞", "量价齐飞已过期")),
|
|
("vp_fly_1h_current", ("量价齐飞", "量价齐飞K")),
|
|
("short_tf_resonance", ("短周期共振", "5m/15m共振")),
|
|
("short_tf_15m_ignition", ("15min短周期启动", "15m短周期启动", "15min 早期启动")),
|
|
("short_tf_5m_ignition", ("5min极早期启动", "5m极早期启动", "5min 早期启动")),
|
|
("volume_divergence_1h", ("量价背离", "放量但无量价齐飞")),
|
|
("static_accum_4h", ("静K蓄力", "静K旁路")),
|
|
("box_breakout_pullback_1h", ("1H", "箱体", "突破", "回踩")),
|
|
("box_breakout_pullback_4h", ("4H", "箱体", "突破", "回踩")),
|
|
("higher_lows_4h", ("底部抬高",)),
|
|
("compression_surge_4h", ("压缩放量",)),
|
|
("ignition_stale", ("历史起爆点", "起爆点已过期", "旧起爆")),
|
|
("ignition_d1_current", ("日线", "起爆点")),
|
|
("ignition_4h_current", ("4H", "起爆点")),
|
|
("ignition_1h_current", ("1H", "起爆点")),
|
|
("dynamic_k_d1_bull", ("日线", "动K")),
|
|
("dynamic_k_1h_bull", ("1H", "动K")),
|
|
("breakout_pullback_d1", ("日线", "突破", "回踩")),
|
|
("breakout_15m_current", ("15min", "突破")),
|
|
("pullback_15m_confirm", ("15min", "回踩确认")),
|
|
("strong_resonance_bypass", ("强共振旁路",)),
|
|
("entry_quality_gate", ("买点质量闸门",)),
|
|
("top_trader_long", ("大户偏多",)),
|
|
("sector_rotation", ("板块联动", "龙头")),
|
|
("sentiment_resonance", ("舆情共振",)),
|
|
("dex_volume_spike", ("DEX", "放量")),
|
|
("liquidity_add", ("流动性", "增加")),
|
|
("liquidity_remove_risk", ("流动性", "撤出")),
|
|
("exchange_outflow", ("交易所", "流出")),
|
|
("exchange_inflow_risk", ("交易所", "流入")),
|
|
("whale_accumulation", ("鲸鱼", "增持")),
|
|
("holder_concentration_risk", ("持仓集中",)),
|
|
("smart_money_buying", ("聪明钱", "买入")),
|
|
("funding_extreme", ("资金费率极端",)),
|
|
("trend_exhaustion", ("衰减", "反转", "阴动K")),
|
|
("false_breakout", ("假突破", "冲高回落")),
|
|
("high_position_reject", ("高位", "追高")),
|
|
("risk_reward_bad", ("risk_reward_ok=false", "rr1=", "盈亏比")),
|
|
]
|
|
|
|
|
|
def signal_code(signal: Any) -> str:
|
|
text = str(signal or "").strip()
|
|
if not text:
|
|
return "unknown"
|
|
normalized = re.sub(r"\s+", "", text)
|
|
for code, needles in _PATTERNS:
|
|
if all(str(needle).replace(" ", "") in normalized for needle in needles):
|
|
return code
|
|
return "unknown"
|
|
|
|
|
|
def signal_label_for_code(code: str) -> str:
|
|
return SIGNAL_CODE_LABELS.get(code or "unknown", SIGNAL_CODE_LABELS["unknown"])
|
|
|
|
|
|
def signal_codes(signals: Iterable[Any]) -> list[str]:
|
|
seen = []
|
|
for sig in signals or []:
|
|
code = signal_code(sig)
|
|
if code not in seen:
|
|
seen.append(code)
|
|
return seen
|
|
|
|
|
|
def signal_labels(signals: Iterable[Any]) -> list[str]:
|
|
return [str(sig) for sig in (signals or []) if str(sig).strip()]
|