"""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", }, "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", "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_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_theme: basis.append("主题/板块线索") if stale_only: basis.append("旧信号仅作背景") if has_15m_trigger and has_1h_momentum and not stale_only: 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_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 == "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}