"""缠论多级别联立分析 【本版改动 · 卖点区间套修复 (第24/44课)】 旧逻辑的两个病根(对应回测中"卖在低位"与"提前卖飞"): ① 卖在低位: 严格模式要求"30分钟出现三类卖点S3"才印证日线卖点。但S3是 次级别已经跌破其中枢、反抽不回的【转折确认点】—— 它出现时价格早已 离开顶部一大截。拿S3当卖出闸门, 等于规定"必须先跌下来才准卖", 结构上 注定卖不到一卖的高位, 最后落在二卖/三卖的低位。 ② 提前卖飞: 宽松模式下"30m最后一笔向下"即印证。第24课明示: 小级别背驰 (更别说仅仅一笔向下)未必引发大级别转折。任何一次30m级别的正常回调都 可能把日线WEAK级别的卖点"确认"掉 → 还没到顶就出局。 而且旧代码对次级别卖点【不做时间嵌套检查】: 30m在日线C段开始之前的 旧背驰也会被当作印证 —— "区间套"三个字里的"区间"丢了。 新逻辑 (CFG['sell_nested_interval']=True): 第44课区间套的本义: 大级别进入背驰段后, 在该背驰段【时间区间内】找次级别 的背驰段, 再在其中找次次级别的背驰段, 逐级收敛到精确转折点。 对日线S1/S2, 次级别印证按优先级: ① 嵌套顶背驰: 30m出现S1, 且其C段顶部落在日线背驰段(最后中枢之后→当下) 的时间窗口内、顶部价位贴近日线C段高点 → 精确卖点, 卖在高位区。 ② 次级别转折确认: 30m出现S3, 或30m末笔已跌破其最近中枢下沿ZD(第92课 向下变盘) → 转折已确认, 偏晚但必须卖。 ③ 两者皆无 → 次级别动能未竭, 第24课: 顶未到 → 【不卖】(防卖飞), 返回 action=HOLD + sell_armed=True 进入"区间套布防"状态: 回测端持仓转入布防, 之后任一条件触发(嵌套背驰出现/破30m中枢ZD/ 较布防峰值回落超阈值)即离场 —— 既不提前卖飞, 也不一路坐滑梯到三卖。 S3(日线三卖)不走嵌套背驰: 三卖本身就是转折确认型卖点, 维持原确认逻辑。 """ from __future__ import annotations from dataclasses import dataclass, field from typing import Optional import numpy as np import pandas as pd # from chan_engine import ChanAnalyzer, Signal # ── 卖点区间套参数 (第24/44课) ── NESTED_TOP_PRICE_TOL = 0.03 # 嵌套顶背驰的顶部须贴近父级C段高点(3%以内), 否则属上一波动能 NESTED_WINDOW_PAD_DAYS = 3 # 时间嵌套窗口的左侧容差(日) SELL_ARM_PEAK_DROP = 0.04 # 布防后较峰值价回落≥4% → 顶部确认离场(回测端使用) SELL_ARM_DISARM_BREAK = 0.03 # 布防后强势创新高超3%且卖点消失 → 背驰被消化, 撤防(第26课) def _default_make_analyzer(level, df): return ChanAnalyzer(df.reset_index(drop=True)) _MAKE_ANALYZER = _default_make_analyzer def set_analyzer_factory(fn): global _MAKE_ANALYZER _MAKE_ANALYZER = fn or _default_make_analyzer def resample_weekly(df_daily: pd.DataFrame) -> pd.DataFrame: if df_daily.empty: return df_daily.copy() d = df_daily.copy() d['date'] = pd.to_datetime(d['date']) d = d.sort_values('date').reset_index(drop=True) wk_period = d['date'].dt.to_period('W-FRI') agg = {'open': 'first', 'high': 'max', 'low': 'min', 'close': 'last'} if 'volume' in d.columns: agg['volume'] = 'sum' if 'amount' in d.columns: agg['amount'] = 'sum' g = d.groupby(wk_period, sort=True) wk = g.agg(agg) last_real = g['date'].max() wk = wk.dropna(subset=['open', 'high', 'low', 'close']).reset_index(drop=True) wk['date'] = list(last_real.values) return wk def resample_monthly(df_daily: pd.DataFrame) -> pd.DataFrame: if df_daily.empty: return df_daily.copy() d = df_daily.copy() d['date'] = pd.to_datetime(d['date']) d = d.sort_values('date').reset_index(drop=True) mp = d['date'].dt.to_period('M') agg = {'open': 'first', 'high': 'max', 'low': 'min', 'close': 'last'} if 'volume' in d.columns: agg['volume'] = 'sum' if 'amount' in d.columns: agg['amount'] = 'sum' g = d.groupby(mp, sort=True) mo = g.agg(agg) last_real = g['date'].max() mo = mo.dropna(subset=['open', 'high', 'low', 'close']).reset_index(drop=True) mo['date'] = list(last_real.values) return mo @dataclass class LevelView: level: str trend: str n_pivots: int n_bis: int last_bi_dir: str signal: Optional[Signal] zg: Optional[float] zd: Optional[float] dif: Optional[float] last_date: Optional[pd.Timestamp] diagnostics: dict = field(default_factory=dict) @dataclass class MultiLevelSignal: code: str analysis_date: pd.Timestamp weekly: LevelView daily: LevelView m30: Optional[LevelView] action: str confidence: str final_kind: str cur_price: float chain: list = field(default_factory=list) blocked_reason: str = '' note: str = '' confidence_reasons: list = field(default_factory=list) diagnostics: dict = field(default_factory=dict) monthly: Optional[LevelView] = None # ── 卖点区间套布防 (第44课) ── sell_armed: bool = False # 日线S1/S2背驰已现、次级别动能未竭 → 持仓布防 arm_zd: Optional[float] = None # 布防线①: 30m最近中枢下沿ZD(跌破即次级别转折确认) arm_high: Optional[float] = None # 父级背驰段C段高点(用于"新高消化背驰"撤防判定) def explain(self) -> str: lines = [f' [{self.code}] {self.analysis_date.strftime("%Y-%m-%d")} ¥{self.cur_price:.3f}'] lines.append(f' 最终: {self.action} 置信度={self.confidence}' + (f' 信号={self.final_kind}' if self.final_kind else '')) lines.append(' ── 逐级裁决链 ──') for lvl, concl, src in self.chain: lines.append(f' [{lvl:<7s}] {concl}') if src: lines.append(f' └ 依据: {src}') if self.blocked_reason: lines.append(f' ⚠ 拦截: {self.blocked_reason}') if self.note: lines.append(f' 说明: {self.note}') if self.confidence_reasons: lines.append(' 置信度降档明细:') for i, r in enumerate(self.confidence_reasons, 1): lines.append(f' {i}. {r}') if self.diagnostics: lines.append(' 日线买卖点逐项诊断:') for k in ('B1', 'B2', 'B3', 'S1', 'S2', 'S3'): rows = self.diagnostics.get(k) or [] if not rows: continue ok = all(str(x).startswith('✓') for x in rows) lines.append(f' {k} {"满足" if ok else "不满足"}') for row in rows: lines.append(f' {row}') else: lines.append(' 日线买卖点逐项诊断: 未生成') return '\n'.join(lines) class MultiLevelChan: CFG = { 'use_monthly_gate': True, 'monthly_ma_filter': False, 'require_sublevel_sell_confirm': False, 'mode': 'short', 'zhongyin_block_buy': False, 'expose_m30_nosignal': False, 'require_60m_buy_confirm': False, # L44区间套: 60m为日线直接次级别, 买点须经其印证 # L50 MACD级别选择: 看MACD要选"黄白线和柱子清晰"的级别。30m判据模糊而60m # 清晰且60m已印证时, 采信60m的印证结论(避免被模糊级别的噪声误杀好买点)。 'l50_macd_level_select': False, # ── 卖点区间套(第24/44课): 嵌套顶背驰精确定位 + 布防防卖飞 ── # True = S1/S2用"时间嵌套的次级别顶背驰"定位高点; 未嵌套确认时不卖而布防 # False = 维持旧逻辑(30m要S3 / 宽松末笔向下即卖) 'sell_nested_interval': True, # L24: 周线上涨+日线盘整背驰的S1/S2 → 不全清, 转布防短差(撤防可骑回趋势) 'l24_weekly_uptrend_arm': False, # 实测净亏(好卖点也被转布防), 默认关 } # ── 速度优化: 次级别K线只取最近N根做缠论分解 ── # 第17/44课: 次级别(60m/30m/15m/5m/1m)的职责是"印证当下转折", 其买卖点 # 只取决于最近的笔/线段/中枢结构; 几年前的分钟级历史对当下印证毫无贡献, # 却让每个回测日都对上万根分钟K线重做合并/分型/笔/段, 是最大的耗时点。 # 截断只作用于次级别印证, 日/周/月线仍用全history(年线、月线大方向不受影响)。 SUB_TAIL = {'15m': 4000, '5m': 4800, '1m': 2000} def __init__(self, df_daily, df_weekly=None, df_monthly=None, df_30m=None, df_60m=None, df_15m=None, df_5m=None, df_1m=None, code='', strict=True): self.code = code self.strict = strict self.df_daily = df_daily.reset_index(drop=True) if not df_daily.empty else df_daily self.df_30m = df_30m.reset_index(drop=True) if df_30m is not None and not df_30m.empty else None self.df_60m = df_60m.reset_index(drop=True) if df_60m is not None and not df_60m.empty else None self.df_15m = df_15m.reset_index(drop=True) if df_15m is not None and not df_15m.empty else None self.df_5m = df_5m.reset_index(drop=True) if df_5m is not None and not df_5m.empty else None self.df_1m = df_1m.reset_index(drop=True) if df_1m is not None and not df_1m.empty else None self.df_weekly = df_weekly.reset_index(drop=True) if df_weekly is not None and not df_weekly.empty else None self.df_monthly = df_monthly.reset_index(drop=True) if df_monthly is not None and not df_monthly.empty else None @staticmethod def _make_view(level, df, an=None, diagnose=True): if an is None: if df is None or len(df) < 30: return None try: an = _MAKE_ANALYZER(level, df) except Exception: return None if an is None: return None sig = an.get_signal() zg = an.pivots[-1].zg if an.n_pivots > 0 else None zd = an.pivots[-1].zd if an.n_pivots > 0 else None return LevelView(level=level, trend=an.trend, n_pivots=an.n_pivots, n_bis=an.n_bis, last_bi_dir=an.bis[-1].direction if an.n_bis else '', signal=sig, zg=zg, zd=zd, dif=float(an.dif.iloc[-1]) if len(an.dif) else None, last_date=pd.Timestamp(df['date'].iloc[-1]), diagnostics=(an.diagnose() if diagnose else {})) # ────────────────────────────────────────────────────────────────── # 卖点区间套核心: 时间嵌套的次级别顶背驰 (第44课) # ────────────────────────────────────────────────────────────────── @staticmethod def _parent_sell_window(parent_sig): """父级(日线)背驰段的时间窗口与顶部价。 S1: 背驰段C段 = 最后中枢结束 → 当下; 顶 = C段高点 c_high。 S2: 窗口 = 一卖出现 → 当下(反抽段); 顶 = 一卖高点 prev_high。""" ex = (parent_sig.extras or {}) if parent_sig is not None else {} if parent_sig is None: return '', '', None if parent_sig.kind == 'S1': w_start = ex.get('pivot_end_date') or ex.get('a_high_date') or '' top = ex.get('c_high') else: # S2 w_start = ex.get('s1_date') or ex.get('prev_high_date') or '' top = ex.get('prev_high') w_end = ex.get('price_date') or '' return w_start, w_end, top def _nested_sell_check(self, sub_an, parent_sig, lvl_name, parent_kind): """第44课区间套(卖点版): 在父级背驰段时间窗口内找次级别的嵌套顶背驰。 优先级: ①嵌套S1(精确高点, 卖在高位) ②S3 / 末笔破次级别中枢ZD(转折已确认, 偏晚但必卖) ③都没有 → 第24课: 次级别动能未竭, 顶未到 → 不卖(防卖飞), 转布防。 返回 (confirmed: bool, note: str)。""" w_start, w_end, parent_top = self._parent_sell_window(parent_sig) def in_window(dstr): if not w_start or not dstr: return True # 信息不足时不因窗口否决(保守放行, 由价位贴近度把关) try: d = pd.Timestamp(dstr) lo = pd.Timestamp(w_start) - pd.Timedelta(days=NESTED_WINDOW_PAD_DAYS) hi = (pd.Timestamp(w_end) if w_end else d) + pd.Timedelta(days=1) return lo <= d <= hi except Exception: return True rejected = '' # ① 嵌套顶背驰 S1 —— 区间套的本体: 背驰段中套背驰段 s1 = sub_an.detect_s1() if s1 is not None: ex2 = s1.extras or {} td = ex2.get('c_high_date', '') tp = ex2.get('c_high') near_top = (parent_top is None or tp is None or tp >= parent_top * (1 - NESTED_TOP_PRICE_TOL)) if in_window(td) and near_top: ptxt = f'(顶¥{tp:.3f}@{td})' if tp else '' return True, (f'{lvl_name}嵌套顶背驰S1{ptxt}落在日线{parent_kind}背驰段窗口内 ' f'—— 第44课区间套: 背驰段中套背驰段, 精确定位顶部, 卖在高位区') rejected = (f'{lvl_name}虽有S1但【不嵌套】(顶@{td or "?"}不在父级背驰段' f'[{w_start}~{w_end}]窗口内, 或低于父级顶{NESTED_TOP_PRICE_TOL:.0%}以上)' f' → 属上一波的动能衰竭, 不作本次印证(防卖飞); ') # ①b 父级为S2时, 次级别S2(一卖后反抽确认)也可嵌套印证 if parent_kind == 'S2': s2 = sub_an.detect_s2() if s2 is not None: ex2 = s2.extras or {} td = ex2.get('cur_high_date', '') if in_window(td): return True, (f'{lvl_name}嵌套二卖S2(反抽顶@{td})落在日线S2窗口内 ' f'—— 第14课: 大级别二卖由次级别相应卖点定位') # ② 转折已确认型: S3 / 末笔已破次级别中枢下沿 s3 = sub_an.detect_s3() if s3 is not None: return True, (f'{lvl_name}出现三类卖点S3 —— 第44课: 最后一个次级别中枢出现三卖, ' f'转折已确认(偏晚, 但必须卖)') try: osc = sub_an.zhongshu_oscillation_monitor() if osc.get('alert') and osc.get('direction') == 'down': return True, (f'{lvl_name}末笔已离开其最近中枢下沿(第92课向下变盘) ' f'—— 次级别转折确认, 高位区兑现') except Exception: pass # ③ 次级别上冲动能已竭的当下确认: 末笔向下 + MACD柱已翻负(黄白线收敛下行)。 # 第24/44课: 区间套要的是"次级别走势的当下转折"; 末笔转下且红柱消失, # 说明次级别这一冲已经结束 —— 此时日线背驰卖点立即执行, 不再等更深的破位 # (这正是旧版"等30m三卖"卖在低位的病根)。只有次级别仍在上冲(末笔向上/ # 红柱未消)时才转入布防, 防的才是真正的"卖飞"。 try: last_bi = sub_an.bis[-1] if sub_an.n_bis else None bar_now = float(sub_an.macd_bar.iloc[-1]) if len(sub_an.macd_bar) else 0.0 if last_bi is not None and last_bi.direction == 'down' and bar_now < 0: return True, (f'{lvl_name}末笔向下且MACD柱已翻负 —— 第24课: 次级别上冲动能已竭, ' f'当下转折确认, 日线背驰卖点立即执行(不等{lvl_name}跌出三卖)') except Exception: pass return False, (rejected + f'{lvl_name}动能未竭(末笔仍向上/MACD红柱未消): 无嵌套顶背驰/无S3 ' f'→ 第24课: 次级别未转折, 大级别顶未到 → 暂不卖(防卖飞), 转入区间套布防') def _confirm_30m_for_buy(self, m30, parent_kind=''): if m30 is None or m30.n_bis < 5: return False, '30分钟笔数不足(<5), 无法做次级别精确印证' if parent_kind == 'B2': if m30.detect_b1() is not None: return True, '30分钟出现一类买点B1 —— 第14课: 大级别第二类买点由次一级别相应走势的一类买点构成' if m30.detect_b2() is not None: return True, '30分钟出现二类买点B2 —— 次级别一买后的回试确认, 属延后确认, 精度低于B1' return False, '30分钟未出现B1/B2 —— 大级别二买缺少次级别买点确认' for fn, name in ((m30.detect_b1,'B1'),(m30.detect_b3,'B3'),(m30.detect_b2,'B2')): if fn() is not None: return True, f'30分钟出现标准买点{name} —— 第44课: 次级别走势确认转折, 日线买点成立' if self.strict: return False, '30分钟未出现任何标准买点(B1/B2/B3) —— 严格模式: 第44课次级别印证未满足, 日线买点暂不成立' if m30.bis[-1].direction == 'up': return True, '30分钟最后一笔向上, 次级别已启动(宽松确认)' return False, '30分钟最后一笔仍向下且无买点, 次级别未确认转折' def _confirm_30m_for_sell(self, m30, parent_kind='', parent_sig=None): if m30 is None or m30.n_bis < 5: return False, '30分钟笔数不足(<5), 无法做次级别精确印证' # ── 新: 卖点区间套(第24/44课) —— S1/S2用嵌套顶背驰定位高点 ── if (self.CFG.get('sell_nested_interval') and parent_sig is not None and parent_kind in ('S1', 'S2')): return self._nested_sell_check(m30, parent_sig, '30分钟', parent_kind) # ── 旧逻辑(S3父级 / 开关关闭时) ── if parent_kind == 'S2': if m30.detect_s1() is not None: return True, '30分钟出现一类卖点S1 —— 大级别第二类卖点由次一级别相应走势的一类卖点精确定位' if m30.detect_s2() is not None: return True, '30分钟出现二类卖点S2 —— 次级别一卖后的反抽确认, 属延后确认, 精度低于S1' return False, '30分钟未出现S1/S2 —— 大级别二卖缺少次级别卖点确认' if m30.detect_s3() is not None: return True, '30分钟出现三类卖点S3 —— 严格满足第44课"小背驰-大转折定理"的必要条件(最后一个次级别中枢出现三卖)' if self.strict: return False, '30分钟未出现三类卖点S3 —— 严格模式: 第44课明确要求"最后一个次级别中枢出现三卖", 必要条件未满足, 日线卖点不构成大级别转折' if m30.detect_s1() is not None: return True, '30分钟出现一类卖点(顶背驰), 次级别转折确认(宽松)' if m30.bis[-1].direction == 'down': return True, '30分钟最后一笔向下, 次级别已转弱(宽松确认)' return False, '30分钟未出现卖点且最后一笔仍向上, 次级别未确认转折' def _confirm_finer(self, an, is_buy, lvl_name, parent_name, parent_kind='', parent_sig=None): if an is None or an.n_bis < 5: return False, f'{lvl_name}笔数不足(<5), 无法做次级别精确印证' if is_buy: if parent_kind == 'B2': if an.detect_b1() is not None: return True, f'{lvl_name}出现B1 —— 第14课: {parent_name}二买由次一级别一买构成' if an.detect_b2() is not None: return True, f'{lvl_name}出现B2 —— 次级别一买后的回试确认, 属延后确认, 精度低于B1' return False, f'{lvl_name}未出现B1/B2 —— {parent_name}二买缺少次级别买点确认' for fn, name in ((an.detect_b1,'B1'),(an.detect_b3,'B3'),(an.detect_b2,'B2')): if fn() is not None: return True, f'{lvl_name}出现标准买点{name} —— 第44课逐级处理: {lvl_name}走势确认{parent_name}的转折' if self.strict: return False, f'{lvl_name}未出现标准买点(B1/B2/B3) —— 严格模式: 末级印证未满足, 信号降级' if an.bis[-1].direction == 'up': return True, f'{lvl_name}最后一笔向上, 次级别已启动(宽松确认)' return False, f'{lvl_name}最后一笔仍向下且无买点, 末级未确认, 信号降级' else: # ── 新: 卖点区间套向更细级别逐级传递(同一父级背驰段窗口) ── if (self.CFG.get('sell_nested_interval') and parent_sig is not None and parent_kind in ('S1', 'S2')): ok, note = self._nested_sell_check(an, parent_sig, lvl_name, parent_kind) if ok: return True, note + f' —— 第44课逐级处理: {lvl_name}确认{parent_name}的转折' return False, note if parent_kind == 'S2': if an.detect_s1() is not None: return True, f'{lvl_name}出现S1 —— {parent_name}二卖由次一级别一卖精确定位' if an.detect_s2() is not None: return True, f'{lvl_name}出现S2 —— 次级别一卖后的反抽确认, 属延后确认, 精度低于S1' return False, f'{lvl_name}未出现S1/S2 —— {parent_name}二卖缺少次级别卖点确认' if an.detect_s3() is not None: return True, f'{lvl_name}出现三类卖点S3 —— 第44课逐级处理: {lvl_name}走势确认{parent_name}的转折' if self.strict: return False, f'{lvl_name}未出现三类卖点S3 —— 严格模式: 末级印证未满足, 信号降级' if an.detect_s1() is not None: return True, f'{lvl_name}出现一类卖点(顶背驰), 次级别转折确认(宽松)' if an.bis[-1].direction == 'down': return True, f'{lvl_name}最后一笔向下, 次级别已转弱(宽松确认)' return False, f'{lvl_name}未出现卖点且最后一笔仍向上, 末级未确认, 信号降级' def _confirm_5m(self, m5, is_buy, parent_kind='', parent_sig=None): return self._confirm_finer(m5, is_buy, '5分钟', '30分钟', parent_kind, parent_sig) def _confirm_1m(self, m1, is_buy, parent_kind='', parent_sig=None): return self._confirm_finer(m1, is_buy, '1分钟', '5分钟', parent_kind, parent_sig) def analyze(self, analysis_date=None, positions=None): if self.df_daily.empty or len(self.df_daily) < 30: return None df_d = self.df_daily if analysis_date is not None: analysis_date = pd.Timestamp(analysis_date) df_d = df_d[df_d['date'] <= analysis_date].reset_index(drop=True) if len(df_d) < 30: return None last_date = pd.Timestamp(df_d['date'].iloc[-1]) cur_price = float(df_d['close'].iloc[-1]) if self.df_weekly is not None: df_w = self.df_weekly[self.df_weekly['date'] <= last_date].reset_index(drop=True) else: df_w = resample_weekly(df_d) if self.df_monthly is not None: df_mo = self.df_monthly[self.df_monthly['date'] <= last_date].reset_index(drop=True) else: df_mo = resample_monthly(df_d) _tail = self.SUB_TAIL or {} def _cut_sub(df, lvl): if df is None: return None sub = df[df['date'] <= last_date + pd.Timedelta(days=1)] n = _tail.get(lvl, 0) if n and len(sub) > n: sub = sub.tail(n) return sub.reset_index(drop=True) df_6 = _cut_sub(self.df_60m, '60m') df_m = _cut_sub(self.df_30m, '30m') df_15 = _cut_sub(self.df_15m, '15m') df_5 = _cut_sub(self.df_5m, '5m') df_1 = _cut_sub(self.df_1m, '1m') wv = self._make_view('weekly', df_w, diagnose=False) mov = self._make_view('monthly', df_mo, diagnose=False) if self.CFG.get('use_monthly_gate') else None daily_an = None try: if len(df_d) >= 30: daily_an = _MAKE_ANALYZER('daily', df_d) except Exception: daily_an = None dv = self._make_view('daily', df_d, an=daily_an, diagnose=False) if daily_an is not None \ else self._make_view('daily', df_d, diagnose=False) m60_an = None m60_built = False def _get_m60(): nonlocal m60_an, m60_built if not m60_built: m60_built = True if df_6 is not None and len(df_6) >= 30: try: m60_an = _MAKE_ANALYZER('60m', df_6) except Exception: m60_an = None return m60_an m30_an = None m30_built = False def _get_m30(): nonlocal m30_an, m30_built if not m30_built: m30_built = True if df_m is not None and len(df_m) >= 30: try: m30_an = _MAKE_ANALYZER('30m', df_m) except Exception: m30_an = None return m30_an def _mv(): an = _get_m30() if an is None: return None return self._make_view('30m', df_m, an=an, diagnose=False) diagnostics = daily_an.diagnose() if daily_an is not None else {} chain = [] yearline_dir = 'unknown'; yearline_val = None if len(df_d) >= 60: _n = min(250, len(df_d)) yearline_val = float(df_d['close'].tail(_n).mean()) if _n >= 20 else None _ma = df_d['close'].rolling(min(250, len(df_d)), min_periods=20).mean() if len(_ma) and not pd.isna(_ma.iloc[-1]): yearline_val = float(_ma.iloc[-1]) if yearline_val is not None: above = cur_price >= yearline_val yearline_dir = 'above' if above else 'below' ytxt = (f'现价¥{cur_price:.3f} {"≥" if above else "<"} 年线MA250 ¥{yearline_val:.3f} ' f'→ {"站上年线, 长线可做多" if above else "年线下方, 第106课不做多(只看反弹/卖点)"}') chain.append(('年线', ytxt, '第7/106课: 年线(MA250)是长线生命线; 站上才考虑做多, 跌破年线长线转空')) else: chain.append(('年线', '日线历史不足, 年线MA250 暂不可用', '第7/106课')) monthly_dir = 'unknown' if mov is not None: monthly_dir = mov.trend chain.append(('monthly', f'{monthly_dir} (月线定大方向: L69 月线看最实质大方向)', '第69/108课: 月线分型/笔/线段确定中期大方向; 底部=第一类买点到中枢首次走出三买前')) if self.CFG.get('monthly_ma_filter') and len(df_mo) >= 5: ma5_m = float(df_mo['close'].tail(5).mean()) if cur_price < ma5_m and monthly_dir != 'down_trend': monthly_dir = 'down_trend' chain.append(('monthly', f'价({cur_price:.2f})在5月线({ma5_m:.2f})下 → 大方向按偏空处理', '第106课: 5月线是长线的关键, 牛市第一轮调整不跌破5月线')) if wv is None: chain.append(('weekly', '周线数据不足(<30根), 方向未知', '')) weekly_dir = 'unknown' else: weekly_dir = wv.trend dir_txt = {'up_trend':'上涨趋势 → 日线只接受【买点】','down_trend':'下跌趋势 → 日线只接受【卖点】/观望', 'consolidation':'盘整 → 日线买卖点都看,但降级'}.get(weekly_dir, weekly_dir) chain.append(('weekly', f'{weekly_dir} {dir_txt}', '第43课: 大级别走势类型限定小级别的操作方向; 不允许"上涨+上涨""下跌+下跌"')) if dv is None: return None daily_sig = dv.signal if daily_sig is None: chain.append(('daily', f'{dv.trend}, 无买卖点信号', '')) else: chain.append(('daily', f'出现 {daily_sig.kind}: {daily_sig.reason[:400]}', '第21课: 买卖点完备性定理')) if daily_sig is None: action, conf, note = self._no_signal_decision(wv, dv, cur_price) m30_ns = _mv() if self.CFG.get('expose_m30_nosignal') else None return MultiLevelSignal(code=self.code, analysis_date=last_date, weekly=wv, daily=dv, m30=m30_ns, action=action, confidence=conf, final_kind='', cur_price=cur_price, chain=chain, note=note, diagnostics=diagnostics, monthly=mov) is_buy = daily_sig.kind in ('B1','B2','B3') is_sell = daily_sig.kind in ('S1','S2','S3') if self.CFG.get('zhongyin_block_buy') and is_buy and daily_sig.kind != 'B3' and daily_an is not None: zy = daily_an.in_zhongyin() if zy.get('in_zhongyin'): chain.append(('中阴', f'第88-90课: 当前处中阴(方向未定+BOLL收口) → 暂不开新仓: {zy["reason"][:120]}', '第89课: 中阴阶段方向未定, 不宜开新仓')) return MultiLevelSignal(code=self.code, analysis_date=last_date, weekly=wv, daily=dv, m30=None, action='WATCH', confidence='LOW', final_kind=daily_sig.kind, cur_price=cur_price, chain=chain, blocked_reason='中阴方向未定', note='中阴阶段暂不开仓(L88-90)', diagnostics=diagnostics, monthly=mov) if positions: fail_kinds = [] for bk, pos in positions.items(): stop = pos.get('stop_px') if stop is not None and cur_price < stop: fail_kinds.append(bk) if fail_kinds: chain.append(('证伪', f'持仓{"/".join(fail_kinds)}跌破建仓日锁定止损线 → 清仓', '第13/20课: 买点结构被破坏即证伪')) return MultiLevelSignal(code=self.code, analysis_date=last_date, weekly=wv, daily=dv, m30=_mv(), action='SELL', confidence='HIGH', final_kind='STOP', cur_price=cur_price, chain=chain, note='结构证伪止损, 次日清仓', diagnostics=diagnostics, monthly=mov) blocked = '' downgrade = False if weekly_dir == 'up_trend' and is_sell: downgrade = True chain.append(('联立','周线上涨 + 日线卖点 → 第44课: 小级别(日线)顶背驰未必引发大级别(周线)转折, 卖点降级为"减仓/短差"', '第24课: 日线背驰除非周线同时背驰,否则不制造周线大顶')) elif weekly_dir == 'down_trend' and is_buy: if daily_sig.kind in ('B1','B2'): chain.append(('联立','周线下跌 + 日线一/二买 → 下跌趋势的转折尝试, 必须由30分钟严格印证','第29课: 下跌的转折=上涨或盘整, 由背驰导致')) else: blocked = '周线下跌趋势中出现日线三买 —— 第43课: 三买属上涨结构, 与周线下跌矛盾, 大概率是下跌中继的假三买' elif monthly_dir == 'down_trend' and is_buy and daily_sig.kind == 'B3' and not blocked: blocked = '月线下跌大方向中出现日线三买 —— 第43/69课: 三买属上涨结构, 与月线大方向矛盾, 大概率假三买' elif weekly_dir == 'up_trend' and is_buy: chain.append(('联立','周线上涨 + 日线买点 → 方向一致, 进入30分钟精确定位','第43课: 大小级别同向, 操作最顺')) elif weekly_dir == 'down_trend' and is_sell: chain.append(('联立','周线下跌 + 日线卖点 → 方向一致, 趋势性卖出','第43课: 大小级别同向')) elif weekly_dir == 'consolidation': downgrade = True chain.append(('联立','周线盘整 → 日线信号有效但降级(盘整中买卖点力度弱)','第29课: 盘整中的转折力度弱于趋势')) # ── 周线二买 → 日线一买精确定位锚 (第14/17课) ── if is_buy and wv is not None and wv.signal is not None \ and getattr(wv.signal, 'kind', '') == 'B2': _dex = (daily_sig.extras or {}) if daily_sig is not None else {} _anchor = _dex.get('b1_price') or _dex.get('cur_low') if _anchor: daily_sig.extras = dict(_dex) daily_sig.extras['weekly_b2_daily_b1_anchor'] = float(_anchor) chain.append(('联立', f'✓ 周线二买 + 日线买点共振: 周线B2由日线一买构成(第14课), ' f'定位锚=日线一买位¥{float(_anchor):.3f}, 跌破该锚则周线二买证伪', '第17课: 大级别买点的精确定位要落到次级别的具体买点价位')) else: chain.append(('联立', '周线二买成立但日线尚无可定位的一买锚 → 等日线给出一买价位再重仓', '第14课')) if blocked: return MultiLevelSignal(code=self.code, analysis_date=last_date, weekly=wv, daily=dv, m30=_mv(), action='WATCH', confidence='NONE', final_kind=daily_sig.kind, cur_price=cur_price, chain=chain, blocked_reason=blocked, note='周线方向闸门拦截, 不操作', diagnostics=diagnostics, monthly=mov) m60_an = _get_m60() ok60 = False if m60_an is not None: if is_buy: ok60, note60 = self._confirm_finer(m60_an, True, '60分钟', '日线', daily_sig.kind) else: ok60, note60 = self._confirm_finer(m60_an, False, '60分钟', '日线', daily_sig.kind, parent_sig=daily_sig) chain.append(('60m', ('✓ 次级别确认: ' if ok60 else '✗ 次级别未确认(降级): ')+note60, '第44课: 区间套 —— 日线的转折先由直接次级别60分钟走势确认')) if not ok60: downgrade = True if self.CFG.get('require_60m_buy_confirm') and is_buy and daily_sig.kind in ('B1', 'B2'): return MultiLevelSignal(code=self.code, analysis_date=last_date, weekly=wv, daily=dv, m30=_mv(), action='WATCH', confidence='LOW', final_kind=daily_sig.kind, cur_price=cur_price, chain=chain, blocked_reason='60分钟直接次级别未印证买点(L44区间套)', note=f'日线{daily_sig.kind}买点但60m未现B1/B2 → 等直接次级别印证再动手', diagnostics=diagnostics, monthly=mov) else: if df_6 is not None: chain.append(('60m', '60分钟数据不足(<30根) → 跳过该级印证', '第44课: 区间套逐级确认')) m30_an = _get_m30() if m30_an is None: chain.append(('30m','无30分钟数据 → 缺少次级别精确印证, 信号降级','第44课: 操作级别的买卖点需次级别走势确认')) confirmed = False; confirm_note = '缺30分钟数据'; downgrade = True else: if is_buy: confirmed, confirm_note = self._confirm_30m_for_buy(m30_an, daily_sig.kind) else: confirmed, confirm_note = self._confirm_30m_for_sell(m30_an, daily_sig.kind, parent_sig=daily_sig) chain.append(('30m', ('✓ 次级别确认: ' if confirmed else '✗ 次级别未确认: ')+confirm_note, '第44课: 小背驰-大转折定理(必要条件)')) # ── 60m嵌套印证补位 (第44课区间套, 卖点版) ── # 30m动能未竭但60m(日线的直接次级别)已给出嵌套顶背驰/转折确认 → 采信60m。 # 日线的"次级别"本就是60m; 30m是60m的次级别, 不应让更细级别一票否决直接次级别。 if (self.CFG.get('sell_nested_interval') and is_sell and not confirmed and ok60 and daily_sig.kind in ('S1', 'S2')): confirmed = True confirm_note = '30m动能未竭, 但60m(日线直接次级别)已嵌套印证 → 采信60m(第44课: 逐级处理, 直接次级别优先)' chain.append(('联立', '✓ ' + confirm_note, '第44课区间套')) # ── L50 MACD级别选择: 选黄白线和柱子清晰的级别看MACD ── if (self.CFG.get('l50_macd_level_select') and is_buy and not confirmed and ok60 and m60_an is not None and m30_an is not None): try: c60 = m60_an.macd_clarity(); c30 = m30_an.macd_clarity() if c30['score'] < 0.4 and c60['score'] >= max(0.4, c30['score'] * 1.5): confirmed = True chain.append(('联立', f"✓ L50 MACD级别选择: 30m MACD{c30['label']}(清晰度{c30['score']}) 判据不可靠; " f"60m MACD{c60['label']}(清晰度{c60['score']})且60m已印证买点 → 采信60m结论", '第50课: 看MACD要选黄白线和柱子走势清晰的级别')) except Exception: pass m15_an = None; m15_confirmed = None if confirmed and df_15 is not None and len(df_15) >= 30: try: m15_an = _MAKE_ANALYZER('15m', df_15) except Exception: m15_an = None if not confirmed: pass elif df_15 is None: pass elif m15_an is None: chain.append(('15m','15分钟数据不足(<30根) → 跳过该级印证','第44课: 区间套逐级确认')) else: ok15, note15 = self._confirm_finer(m15_an, is_buy, '15分钟', '30分钟', daily_sig.kind, parent_sig=(daily_sig if is_sell else None)) m15_confirmed = ok15 chain.append(('15m', ('✓ 次级别确认: ' if ok15 else '✗ 次级别未确认(降级,不拦截): ')+note15, '第44课: 逐级处理 —— 30分钟的转折由15分钟走势确认')) if not ok15: downgrade = True m5_an = None; m5_confirmed = None if confirmed and df_5 is not None and len(df_5) >= 30: try: m5_an = _MAKE_ANALYZER('5m', df_5) except Exception: m5_an = None if not confirmed: pass elif df_5 is None: chain.append(('5m','无5分钟数据 → 缺该级逐级印证, 信号降级','第44课: 15分钟的转折需5分钟走势逐级确认')); downgrade = True elif m5_an is None: chain.append(('5m','5分钟数据不足(<30根) → 缺该级印证, 信号降级','第44课: 15分钟的转折需5分钟走势逐级确认')); downgrade = True else: ok5, note5 = self._confirm_5m(m5_an, is_buy, daily_sig.kind, parent_sig=(daily_sig if is_sell else None)) m5_confirmed = ok5 chain.append(('5m', ('✓ 次级别确认: ' if ok5 else '✗ 次级别未确认(降级,不拦截): ')+note5, '第44课: 逐级处理 —— 15分钟的转折由5分钟走势确认')) if not ok5: downgrade = True m1_an = None; m1_confirmed = None do_1m = confirmed and (m5_confirmed is True) if do_1m and df_1 is not None and len(df_1) >= 30: try: m1_an = _MAKE_ANALYZER('1m', df_1) except Exception: m1_an = None if not do_1m: pass elif df_1 is None: chain.append(('1m','无1分钟数据 → 缺区间套最末级印证, 信号降级','第44课: 区间套 —— 5分钟的转折需1分钟走势逐级确认')); downgrade = True elif m1_an is None: chain.append(('1m','1分钟数据不足(<30根) → 缺最末级印证, 信号降级','第44课: 区间套 —— 5分钟的转折需1分钟走势逐级确认')); downgrade = True else: ok1, note1 = self._confirm_1m(m1_an, is_buy, daily_sig.kind, parent_sig=(daily_sig if is_sell else None)) m1_confirmed = ok1 chain.append(('1m', ('✓ 末级确认: ' if ok1 else '✗ 末级未确认(降级,不拦截): ')+note1, '第44课: 区间套 —— 5分钟的转折由1分钟走势确认')) if not ok1: downgrade = True # ── L24规则: 周线上涨 + 日线【盘整背驰】(非趋势背驰)的S1/S2 ── # 第24课: 某级别的背驰导致该级别的转折; 日线背驰若周线未同步背驰, # 只构成日线级别的调整, 不是大顶 → 不全清仓, 转入区间套布防做短差: # 真转折由布防线(嵌套背驰/破30m中枢ZD/峰值回落)兜底, 趋势延续则由 # 撤防机制(强势新高消化背驰, 第26课)继续骑中枢上移(第91课①)。 if (self.CFG.get('sell_nested_interval') and self.CFG.get('l24_weekly_uptrend_arm') and is_sell and daily_sig.kind in ('S1', 'S2') and weekly_dir == 'up_trend'): _dg = getattr(daily_sig, 'diverge_grade', None) if _dg is not None and not getattr(_dg, 'is_trend_divergence', False): _arm_zd = (m30_an.pivots[-1].zd if (m30_an is not None and m30_an.n_pivots) else None) _ex = daily_sig.extras or {} _arm_high = _ex.get('c_high') or _ex.get('prev_high') or cur_price chain.append(('L24', f'周线上涨 + 日线{daily_sig.kind}仅为盘整背驰(非趋势背驰) ' f'→ 第24课: 不构成周线级别大顶, 不全清 → 转区间套布防短差' f'(布防线: 30m中枢ZD{f"¥{_arm_zd:.3f}" if _arm_zd else "暂无"}/' f'峰值回落{SELL_ARM_PEAK_DROP:.0%}; 强势新高则撤防骑趋势)', '第24课: 日线背驰除非周线同步背驰, 否则只造成日线级别调整')) return MultiLevelSignal(code=self.code, analysis_date=last_date, weekly=wv, daily=dv, m30=_mv(), action='HOLD', confidence='MEDIUM', final_kind=daily_sig.kind, cur_price=cur_price, chain=chain, note=f'L24: 周线上涨中的日线盘整背驰{daily_sig.kind}, 布防短差不全清', sell_armed=True, arm_zd=_arm_zd, arm_high=float(_arm_high) if _arm_high else None, diagnostics=diagnostics, monthly=mov) action = 'BUY' if is_buy else 'SELL' if self.CFG.get('mode') == 'long' and is_sell and daily_sig.kind in ('S1', 'S2'): big_up_long = ((wv is not None and wv.trend == 'up_trend') or (mov is not None and monthly_dir == 'up_trend')) ma5m_ok = True if len(df_mo) >= 5: ma5m = float(df_mo['close'].tail(5).mean()) ma5m_ok = cur_price >= ma5m if big_up_long and ma5m_ok: chain.append(('mode', f'长线模式: 日线{daily_sig.kind}非周线级别转折且大方向上涨 → 持有(操作级别=周线)', '第72课: 看周线则日线调整不构成卖段; 第61课大级别不因小级别震荡卖')) # 长线模式持有也挂上布防线(第44课): 真转折时不至于全程坐滑梯 _arm_zd = (m30_an.pivots[-1].zd if (m30_an is not None and m30_an.n_pivots) else None) _ex = daily_sig.extras or {} _arm_high = _ex.get('c_high') or _ex.get('prev_high') or cur_price return MultiLevelSignal(code=self.code, analysis_date=last_date, weekly=wv, daily=dv, m30=_mv(), action='HOLD', confidence='MEDIUM', final_kind=daily_sig.kind, cur_price=cur_price, chain=chain, note=f'长线模式持有: 日线{daily_sig.kind}属次级别调整, 周线方向未转', sell_armed=bool(self.CFG.get('sell_nested_interval')), arm_zd=_arm_zd, arm_high=float(_arm_high) if _arm_high else None, diagnostics=diagnostics, monthly=mov) if not confirmed: if is_sell: # ── 卖点区间套布防 (第24/44课): 背驰已现但次级别动能未竭 ── # 不立即卖(防卖飞), 也绝不傻等到三卖(防坐滑梯): 持仓转入布防, # 由回测端的布防线(嵌套背驰出现/破30m中枢ZD/峰值回落阈值)收割。 # 布防仅用于S1的精确定顶。S2是一卖后的【反抽】卖点(第15课): # 上方空间被一卖高点封死, 反抽本身就是用来卖的, "等次级别动能衰竭" # 与其性质矛盾 → S2未印证时维持"先卖再说"(走下方LOW置信卖出)。 if (self.CFG.get('sell_nested_interval') and daily_sig.kind == 'S1'): arm_zd = (m30_an.pivots[-1].zd if (m30_an is not None and m30_an.n_pivots) else None) ex = daily_sig.extras or {} arm_high = ex.get('c_high') or ex.get('prev_high') or cur_price chain.append(('卖点布防', f'日线{daily_sig.kind}背驰已现但次级别动能未竭 → 不立即卖(防卖飞), ' f'转入区间套布防: ①次级别出现嵌套顶背驰即卖 ' f'②跌破30m最近中枢下沿ZD{f"¥{arm_zd:.3f}" if arm_zd else "(暂无30m中枢)"}即卖 ' f'③较布防后峰值回落{SELL_ARM_PEAK_DROP:.0%}即卖', '第44课区间套: 背驰段中套背驰段定精确高点; 第24课: 次级别未背驰, 大级别顶未到')) return MultiLevelSignal(code=self.code, analysis_date=last_date, weekly=wv, daily=dv, m30=_mv(), action='HOLD', confidence='MEDIUM', final_kind=daily_sig.kind, cur_price=cur_price, chain=chain, note=f'区间套布防中: 日线{daily_sig.kind}背驰段内, 等次级别嵌套背驰收敛后高位兑现(第44课)', sell_armed=True, arm_zd=arm_zd, arm_high=float(arm_high) if arm_high else None, confidence_reasons=[f'布防原因: {confirm_note}'], diagnostics=diagnostics, monthly=mov) big_up_sell = ((wv is not None and wv.trend == 'up_trend') or (mov is not None and monthly_dir == 'up_trend')) if self.CFG.get('require_sublevel_sell_confirm') and big_up_sell: chain.append(('sell_gate', f'日线{daily_sig.kind}卖点但次级别未印证, 且大方向上涨 → 持有(L61: 大级别不因小级别震荡卖)', '第61课: 区间套确认; 大级别操作不因小级别震荡清仓')) return MultiLevelSignal(code=self.code, analysis_date=last_date, weekly=wv, daily=dv, m30=_mv(), action='HOLD', confidence='LOW', final_kind=daily_sig.kind, cur_price=cur_price, chain=chain, note=f'日线{daily_sig.kind}卖点次级别未印证+大方向上涨, 持有等确认: {confirm_note}', confidence_reasons=[f'次级别未印证持有: {confirm_note}'], diagnostics=diagnostics, monthly=mov) chain.append(('sell_gate',f'日线{daily_sig.kind}卖点已成立, 次级别未印证只降级不拦截','第14/15/21课: 买点买、卖点卖; 区间套用于精确定位, 不是否决卖点')) return MultiLevelSignal(code=self.code, analysis_date=last_date, weekly=wv, daily=dv, m30=_mv(), action='SELL', confidence='LOW', final_kind=daily_sig.kind, cur_price=cur_price, chain=chain, note=f'日线{daily_sig.kind}卖点先执行; 次级别未印证仅降级: {confirm_note}', confidence_reasons=[f'次级别未印证: {confirm_note}'], diagnostics=diagnostics, monthly=mov) return MultiLevelSignal(code=self.code, analysis_date=last_date, weekly=wv, daily=dv, m30=_mv(), action='WATCH', confidence='LOW', final_kind=daily_sig.kind, cur_price=cur_price, chain=chain, blocked_reason=f'次级别未印证({confirm_note})', note='日线有买/非S1卖信号但次级别未确认 -- 等次级别出信号再动手', diagnostics=diagnostics, monthly=mov) score = 100; conf_reasons = [] def _penalty(pts, reason): nonlocal score score -= pts; conf_reasons.append(f'(-{pts}) {reason}') if mov is not None and monthly_dir == 'down_trend' and is_buy: _penalty(25, '第69课: 月线大方向下跌, 日线买点属逆大方向的转折尝试') trend_piv_lt2 = (daily_an is not None and daily_an.n_trend_pivots < 2) if trend_piv_lt2 and daily_sig.kind in ('B1', 'S1'): _penalty(20, f'第27课: 背驰段前仅{daily_an.n_trend_pivots}个中枢(<2), 盘整背驰非趋势背驰') if daily_an is not None and daily_an.trend == 'consolidation': _penalty(15, '第29课: 日线处盘整结构, 盘整中转折力度弱') daily_dg = getattr(daily_sig, 'diverge_grade', None) if daily_dg is not None and daily_dg.grade == 'WEAK': trig = '面积' if daily_dg.area_ok else 'DIF极值' _penalty(15, f'第27课: 背驰仅满足{trig}单一判据(WEAK), 非标准背驰') if (daily_dg is not None and daily_dg.grade in ('STRONG', 'WEAK') and not daily_dg.is_trend_divergence and not trend_piv_lt2 and daily_sig.kind in ('B1', 'S1')): _penalty(10, '第27课: 背驰段为盘整背驰(非趋势背驰), 力度偏弱') if downgrade: _penalty(12, '第44课: 区间套次级别未完全印证(精度降级, 非证伪)') if daily_sig.kind in ('B2', 'S2'): has_delayed = any(lvl in ('30m', '15m', '5m', '1m') and '延后确认' in concl for lvl, concl, _ in chain) if has_delayed: _penalty(8, '二/二卖由次级别延后确认(非次级别一买/一卖精确定位), 精度略降') ex = daily_sig.extras or {} if daily_sig.kind == 'B3': missing = [] if ex.get('b1_price') is None: missing.append('一买') if ex.get('b2_price') is None: missing.append('二买') if missing: _penalty(6, '第20/21课: 三买未能定位对应' + '/'.join(missing) + ', 质量略降') if ex.get('late_trend_b3'): _penalty(12, '第20/92课: 第二个以上同向中枢后的三买, 操作意义下降') if daily_sig.kind == 'S3': missing = [] if ex.get('s1_price') is None: missing.append('一卖') if ex.get('s2_price') is None: missing.append('二卖') if missing: _penalty(6, '第20/21课: 三卖未能定位对应' + '/'.join(missing) + ', 质量略降') if daily_dg is not None and daily_dg.grade == 'STRONG' and daily_dg.is_trend_divergence: score += 10; conf_reasons.append('(+10) 第27课: 标准趋势背驰(>=2中枢+面积+DIF), 高质量') # ── 卖点区间套加分: 30m嵌套顶背驰精确定位(卖在高位区) ── if is_sell and any(lvl == '30m' and '嵌套顶背驰' in concl for lvl, concl, _ in chain): score += 8; conf_reasons.append('(+8) 第44课: 30m嵌套顶背驰精确定位, 卖点落在高位区') score = max(0, min(110, score)) confidence = 'HIGH' if score >= 70 else ('MEDIUM' if score >= 45 else 'LOW') if conf_reasons: chain.append(('置信度', f'{confidence}(评分{score}) ← ' + '; '.join(conf_reasons), '第21课: 一二三类买卖点是位置分类非质量排序; 置信度按原著力度条件评分')) else: chain.append(('置信度', f'HIGH(评分{score}) ← 力度条件全部满足', '第27/29/43课: 方向一致+趋势背驰+趋势结构')) note_parts = [] if conf_reasons: note_parts.append(f'置信评分明细: ' + '; '.join(conf_reasons)) n_lvl = 3 + (self.df_60m is not None) + (self.df_15m is not None) + (self.df_5m is not None) + (self.df_1m is not None) lvl_txt = {3:'三级别',4:'四级别',5:'五级别',6:'六级别',7:'七级别'}.get(n_lvl, f'{n_lvl}级别') m5_txt = ('/5m已印证' if m5_confirmed is True else '/5m未印证(已降级)' if m5_confirmed is False else '') m1_txt = ('/1m已印证' if m1_confirmed is True else '/1m未印证(已降级)' if m1_confirmed is False else '') note_parts.append(f'{lvl_txt}联立通过: 周线{weekly_dir}/日线{daily_sig.kind}/30m已印证{m5_txt}{m1_txt}') note = ' | '.join(note_parts) return MultiLevelSignal(code=self.code, analysis_date=last_date, weekly=wv, daily=dv, m30=_mv(), action=action, confidence=confidence, final_kind=daily_sig.kind, cur_price=cur_price, chain=chain, note=note, confidence_reasons=conf_reasons, diagnostics=diagnostics, monthly=mov) @staticmethod def _no_signal_decision(wv, dv, cur_price): if wv is not None and wv.trend == 'up_trend' and dv.trend == 'up_trend': if dv.zg is not None and cur_price > dv.zg: return ('HOLD','NONE', f'周线+日线均上涨且价({cur_price:.2f})在日线ZG({dv.zg:.2f})上, 继续持有等卖点(第108课: 买点入,持有等卖点)') return ('HOLD','NONE','周线日线均上涨, 趋势中持有') if wv is not None and wv.trend == 'down_trend': return ('WATCH','NONE','周线下跌趋势, 无日线买点 → 空仓观望, 不抄底') return ('WATCH','NONE', f'无明确信号(周线{wv.trend if wv else "?"}/日线{dv.trend}), 观望')