226 lines
8.2 KiB
Python
226 lines
8.2 KiB
Python
"""Opportunity level classification and level-aware trade plan helpers.
|
|
|
|
The recommendation chain should tell users what kind of move it is trying to
|
|
capture. This module keeps that taxonomy stable and keeps level-specific
|
|
entry/SL/TP assumptions out of UI code.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import statistics
|
|
from typing import Any, Dict, Iterable, List, Tuple
|
|
|
|
|
|
OPPORTUNITY_LEVELS: Dict[str, Dict[str, str]] = {
|
|
"intraday_breakout": {
|
|
"label": "日内启动",
|
|
"holding_horizon": "数小时-1天",
|
|
"entry_model": "15m触发 / 1H突破延续",
|
|
"stop_model": "15m摆动低点 / 1H ATR紧止损",
|
|
"tp_model": "1H压力位 / 2-3.5ATR / 移动止盈",
|
|
"max_action": "buy_now",
|
|
},
|
|
"short_swing": {
|
|
"label": "短波段",
|
|
"holding_horizon": "1-3天",
|
|
"entry_model": "1H回踩 / 30m确认",
|
|
"stop_model": "1H摆动低点 / 4H需求区",
|
|
"tp_model": "4H压力位 / 前高 / 移动止盈",
|
|
"max_action": "buy_now",
|
|
},
|
|
"momentum_watch": {
|
|
"label": "强势观察",
|
|
"holding_horizon": "数小时-2天",
|
|
"entry_model": "涨幅榜强势 / 等二次结构",
|
|
"stop_model": "等待回踩低点 / 新结构失效位",
|
|
"tp_model": "不追首段 / 只跟踪二次买点",
|
|
"max_action": "observe",
|
|
},
|
|
"structure_watch": {
|
|
"label": "结构观察",
|
|
"holding_horizon": "3-7天",
|
|
"entry_model": "4H结构回踩后再确认",
|
|
"stop_model": "4H结构低点 / 日线需求区",
|
|
"tp_model": "日线压力位 / 波段高点",
|
|
"max_action": "wait_pullback",
|
|
},
|
|
"theme_trend": {
|
|
"label": "主题趋势",
|
|
"holding_horizon": "1-3周",
|
|
"entry_model": "日线/4H回踩确认",
|
|
"stop_model": "日线趋势失效位",
|
|
"tp_model": "趋势跟踪 / 分批兑现",
|
|
"max_action": "observe",
|
|
},
|
|
}
|
|
|
|
|
|
LEVEL_ORDER = ("intraday_breakout", "short_swing", "momentum_watch", "structure_watch", "theme_trend")
|
|
|
|
|
|
def _text(signals: Iterable[Any]) -> str:
|
|
return " ".join(str(s) for s in (signals or []))
|
|
|
|
|
|
def _has_any(text: str, keywords: Iterable[str]) -> bool:
|
|
return any(k in text for k in keywords)
|
|
|
|
|
|
def _safe_float(value: Any, default: float = 0.0) -> float:
|
|
try:
|
|
if value is None or value == "":
|
|
return default
|
|
return float(value)
|
|
except Exception:
|
|
return default
|
|
|
|
|
|
def _current_trigger(signals: Iterable[Any], entry_plan: Dict[str, Any] | None = None) -> bool:
|
|
plan = entry_plan or {}
|
|
if plan.get("entry_trigger_confirmed") is True:
|
|
return True
|
|
text = _text(signals)
|
|
summary = str(plan.get("pa_15min_summary") or "")
|
|
return (
|
|
"15min即刻入场" in text
|
|
or "当前15min" in text
|
|
or "当前 15min" in text
|
|
or ("即刻入场" in summary and "无动K突破" not in summary)
|
|
)
|
|
|
|
|
|
def opportunity_level_meta(level: str) -> Dict[str, str]:
|
|
level = level if level in OPPORTUNITY_LEVELS else "structure_watch"
|
|
return {"opportunity_level": level, **OPPORTUNITY_LEVELS[level]}
|
|
|
|
|
|
def classify_opportunity_level(
|
|
*,
|
|
signals: Iterable[Any],
|
|
entry_plan: Dict[str, Any] | None = None,
|
|
market_context: Dict[str, Any] | None = None,
|
|
derivatives_context: Dict[str, Any] | None = None,
|
|
sector_context: Dict[str, Any] | None = None,
|
|
m30_aligned: bool = False,
|
|
) -> Dict[str, Any]:
|
|
"""Classify the move level from structured signals and context.
|
|
|
|
The classifier is intentionally conservative: lower timeframe current
|
|
triggers make a move tradable; higher timeframe/theme evidence without a
|
|
fresh entry trigger stays in watch-oriented levels.
|
|
"""
|
|
|
|
plan = entry_plan or {}
|
|
text = _text(signals)
|
|
market_context = market_context or {}
|
|
sector_context = sector_context or {}
|
|
|
|
has_15m_trigger = _current_trigger(signals, plan)
|
|
has_1h_momentum = _has_any(text, ("1H 量价齐飞", "1H放量突破", "1H极放量", "1H 起爆", "1H 动K", "1H 1根量价齐飞"))
|
|
has_30m_bridge = bool(m30_aligned) or _has_any(text, ("30min", "30m"))
|
|
has_4h_or_daily = _has_any(text, ("4H", "日线", "周线", "需求区", "突破回踩", "底部", "静K", "蓄力"))
|
|
has_top_gainer = _has_any(text, ("24h强势榜", "强势榜异动", "涨幅榜"))
|
|
has_chase_risk = _has_any(text, ("追高", "离突破位+", "站稳突破位+"))
|
|
has_theme = bool(sector_context.get("hot_sectors")) or _has_any(
|
|
text,
|
|
("主题", "生态", "舆情", "板块", "listing", "公告", "催化"),
|
|
)
|
|
stale_only = _has_any(text, ("历史", "已过期")) and not has_15m_trigger
|
|
|
|
basis: List[str] = []
|
|
if has_15m_trigger:
|
|
basis.append("当前15m触发")
|
|
if has_1h_momentum:
|
|
basis.append("1H动能确认")
|
|
if has_30m_bridge:
|
|
basis.append("30m结构桥接")
|
|
if has_4h_or_daily:
|
|
basis.append("高周期结构背景")
|
|
if has_top_gainer:
|
|
basis.append("24h强势榜异动")
|
|
if has_chase_risk:
|
|
basis.append("追高/首段已启动风险")
|
|
if has_theme:
|
|
basis.append("主题/板块线索")
|
|
if stale_only:
|
|
basis.append("旧信号仅作背景")
|
|
|
|
if has_15m_trigger and has_1h_momentum and not stale_only and not has_chase_risk:
|
|
level = "intraday_breakout"
|
|
elif (has_1h_momentum and (has_30m_bridge or has_4h_or_daily)) and not stale_only:
|
|
level = "short_swing"
|
|
elif has_top_gainer and (has_15m_trigger or has_4h_or_daily or has_1h_momentum):
|
|
level = "momentum_watch"
|
|
elif has_theme and not has_15m_trigger and not has_1h_momentum:
|
|
level = "theme_trend"
|
|
elif has_4h_or_daily or stale_only:
|
|
level = "structure_watch"
|
|
elif has_1h_momentum:
|
|
level = "short_swing"
|
|
else:
|
|
level = "structure_watch"
|
|
|
|
meta = opportunity_level_meta(level)
|
|
meta["plan_basis"] = basis or ["异动候选进入确认"]
|
|
return meta
|
|
|
|
|
|
def attach_opportunity_level(entry_plan: Dict[str, Any], level_meta: Dict[str, Any]) -> Dict[str, Any]:
|
|
plan = dict(entry_plan or {})
|
|
level = level_meta.get("opportunity_level") or "structure_watch"
|
|
meta = opportunity_level_meta(level)
|
|
plan.update(
|
|
{
|
|
"opportunity_level": level,
|
|
"opportunity_level_label": level_meta.get("label") or meta["label"],
|
|
"holding_horizon": level_meta.get("holding_horizon") or meta["holding_horizon"],
|
|
"entry_model": level_meta.get("entry_model") or meta["entry_model"],
|
|
"stop_model": level_meta.get("stop_model") or meta["stop_model"],
|
|
"tp_model": level_meta.get("tp_model") or meta["tp_model"],
|
|
"max_action": level_meta.get("max_action") or meta["max_action"],
|
|
"plan_basis": list(level_meta.get("plan_basis") or []),
|
|
}
|
|
)
|
|
return plan
|
|
|
|
|
|
def select_level_stop_loss(
|
|
*,
|
|
level: str,
|
|
price: float,
|
|
entry_price: float,
|
|
stop_candidates: Iterable[float],
|
|
) -> Tuple[float, str]:
|
|
"""Pick a stop from candidates according to opportunity level."""
|
|
|
|
ref = _safe_float(entry_price) or _safe_float(price)
|
|
candidates = sorted(
|
|
{
|
|
round(_safe_float(c), 8)
|
|
for c in (stop_candidates or [])
|
|
if _safe_float(c) > 0 and _safe_float(c) < ref
|
|
}
|
|
)
|
|
if not candidates:
|
|
return 0.0, OPPORTUNITY_LEVELS.get(level, OPPORTUNITY_LEVELS["structure_watch"])["stop_model"]
|
|
|
|
if level == "intraday_breakout":
|
|
stop = candidates[-1]
|
|
elif level == "short_swing":
|
|
stop = statistics.median(candidates)
|
|
else:
|
|
stop = candidates[0]
|
|
return round(float(stop), 6), OPPORTUNITY_LEVELS.get(level, OPPORTUNITY_LEVELS["structure_watch"])["stop_model"]
|
|
|
|
|
|
def level_tp_parameters(level: str) -> Dict[str, float]:
|
|
if level == "intraday_breakout":
|
|
return {"tp1_atr": 2.0, "tp1_floor": 0.03, "tp2_atr": 3.5, "tp2_floor": 0.05}
|
|
if level == "short_swing":
|
|
return {"tp1_atr": 3.0, "tp1_floor": 0.05, "tp2_atr": 5.0, "tp2_floor": 0.08}
|
|
if level == "momentum_watch":
|
|
return {"tp1_atr": 3.0, "tp1_floor": 0.06, "tp2_atr": 5.0, "tp2_floor": 0.10}
|
|
if level == "theme_trend":
|
|
return {"tp1_atr": 6.0, "tp1_floor": 0.12, "tp2_atr": 10.0, "tp2_floor": 0.20}
|
|
return {"tp1_atr": 4.0, "tp1_floor": 0.08, "tp2_atr": 7.0, "tp2_floor": 0.14}
|