From 6396d6fe8639f7ebe06aff6fa71e7d9f6e577487 Mon Sep 17 00:00:00 2001 From: xxm Date: Thu, 20 Aug 2026 17:33:04 +0800 Subject: [PATCH] =?UTF-8?q?refactor:=20=E5=BD=92=E6=A1=A3stale=5Fpush=5Fwl?= =?UTF-8?q?in(price=5Fmonitor=E5=B7=B2=E8=A6=86=E7=9B=96=E5=85=B6=E5=8A=9F?= =?UTF-8?q?=E8=83=BD)+=E6=B8=85=E7=90=86=E6=AE=8B=E7=95=99=E5=BC=95?= =?UTF-8?q?=E7=94=A8+=E7=A6=81=E7=94=A8cron=20job?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - stale_push_wlin归档(price_monitor每2分钟监控所有持仓+自选的买入/止损/止盈区间,已完全覆盖) - market_config/mofin_health/signal_ledger注释引用更新 - jobs.json禁用stale_push_wlin cron job --- deploy/profile-scripts/market_config.py | 4 +- deploy/profile-scripts/mofin_health.py | 6 +- deploy/profile-scripts/signal_ledger.py | 2 +- deploy/profile-scripts/stale_push_wlin.py | 1084 --------------------- 4 files changed, 6 insertions(+), 1090 deletions(-) delete mode 100644 deploy/profile-scripts/stale_push_wlin.py diff --git a/deploy/profile-scripts/market_config.py b/deploy/profile-scripts/market_config.py index e50f483a..c34db204 100644 --- a/deploy/profile-scripts/market_config.py +++ b/deploy/profile-scripts/market_config.py @@ -3,7 +3,7 @@ """market_config.py — 市场抽象层（阶段1：港股接入核心结构件） 市场判断 / 行情符号 / 市场配置的**唯一事实源**，收敛散落在 -market_data / price_monitor / stale_push_wlin / technical_analysis / +market_data / price_monitor / technical_analysis / strategy_lab 中的 6 处重复实现。 铁律： @@ -30,7 +30,7 @@ def market_for_code(code): return 'hk' if is_hk_stock(code) else 'a' -# ── 行情符号映射（收敛 market_data/price_monitor/stale_push_wlin/technical_analysis 四处）── +# ── 行情符号映射（收敛 market_data/price_monitor/technical_analysis 三处）── def _normalize_code(code): """归一化代码：去市场前缀（sh/sz/hk/SH...）与 '_' 后缀，返回大写纯代码""" diff --git a/deploy/profile-scripts/mofin_health.py b/deploy/profile-scripts/mofin_health.py index 4b10a86d..dd9e700a 100644 --- a/deploy/profile-scripts/mofin_health.py +++ b/deploy/profile-scripts/mofin_health.py @@ -357,7 +357,7 @@ FLOW_DETAIL = { "import_holding_xls": "从券商holding.xls导入最新持仓数量/成本/市值", }, "readers": { - "stale_push_wlin": "读取持仓列表+最新价格，检查是否进入买入区/触发止损", + "price_monitor": "高频价格监控+操作区间触发重评", "mofin_db": "内部读取（get_price_from_db等函数）", "system_audit": "读取持仓总数/品种分布，审计持仓完整性", "server": "读取持仓数据供Web Dashboard展示", @@ -391,7 +391,7 @@ FLOW_DETAIL = { "data_governance": "读取所有活跃策略，检查缺失和异常", "per_stock_reassess": "读取个股策略配置，判断是否需要重评", "mo_data": "通过read_decisions()读取策略数据", - "stale_push_wlin": "读取买入区/止损/止盈配置，检查价格触发", + "price_monitor": "检测操作区间(买入/止损/止盈)并触发重评", }, }, "live_prices": { @@ -471,7 +471,7 @@ FLOW_DETAIL = { "refresh_macro_context": "每30分钟采集上证/深证/创业板/恒指等指数+情绪指标", }, "readers": { - "stale_push_wlin": "读取大盘情绪用于策略推送的宏观背景", + "market_regime": "温区判断(替代原detect_scenario)", "divergence_detector": "读取多市场指数数据做背离检测", "system_audit": "审计数据采集是否正常", }, diff --git a/deploy/profile-scripts/signal_ledger.py b/deploy/profile-scripts/signal_ledger.py index c62b8063..ce237253 100644 --- a/deploy/profile-scripts/signal_ledger.py +++ b/deploy/profile-scripts/signal_ledger.py @@ -13,7 +13,7 @@ 用法： from signal_ledger import record_signal, get_resonance record_signal(code="300750", name="宁德时代", strategy="v_oversold", version="v_oversold", - regime="trend_down", temp_band="panic", reason="进买入区+重评买入", source_module="stale_push_wlin") + regime="trend_down", temp_band="panic", reason="进买入区+重评买入", source_module="price_monitor") res = get_resonance("300750") # 返回该股票近24h共振信息 """ import json diff --git a/deploy/profile-scripts/stale_push_wlin.py b/deploy/profile-scripts/stale_push_wlin.py deleted file mode 100644 index 89be7ec2..00000000 --- a/deploy/profile-scripts/stale_push_wlin.py +++ /dev/null @@ -1,1084 +0,0 @@ -#!/usr/bin/env python3 -""" -stale_push_wlin.py — 按5步逻辑推送自选股买入区提醒 + 自动触发重评（全DB模式） - -5步逻辑： -1. 筛选 is_watchlist=true 且价在买入区 -2. RR<1.5/无止盈位/非买入signal → 标记 STRATEGY_STALE → 触发自动重评 -3. 可推的：计算每手买入金额和现金占比 -4. 发现 STRATEGY_STALE → 后台跑 per_stock_reassess.py 自动重评 - -所有持仓/策略/现金数据均从DB读取，不再依赖JSON文件。 -宏现上下文和冷却状态仍保留JSON fallback。 -no_agent模式：有推送→输出；无→静默 -搭配 cron: no_agent=True, 交易日每30分跑一次 -""" -import subprocess -import sys, re, json, os, time -import threading -import time -from datetime import datetime, time -from mo_data import read_portfolio, read_decisions, get_price -from mofin_db import get_conn - -# ── MoFin unified model ────────────────────────────────────────────── -sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) -from mo_models import is_hk_stock, get_hk_rate, to_cny, calc_total_assets -from market_config import kline_symbol - -# 市场时段检查 -_MARKET_HOURS = { - 'ashare': (time(9, 30), time(15, 0)), - 'hk': (time(9, 30), time(16, 0)), -} - -def is_ashare(code: str) -> bool: - """判断是否A股代码""" - return code.isdigit() and (code.startswith(('6', '5')) or len(code) in (6,)) - -def market_is_open(code: str, now: datetime = None) -> bool: - """检查某股票对应市场是否在交易时段内""" - if not code: - return True - now = now or datetime.now() - t = now.time() - code_str = str(code) - if code_str.startswith(('0', '1')) and len(code_str) == 5: - # 港股 - start, end = _MARKET_HOURS['hk'] - else: - # A股（含ETF、科创板） - start, end = _MARKET_HOURS['ashare'] - return start <= t <= end -try: - from urllib.request import Request, urlopen -except ImportError: - from urllib2 import Request, urlopen -# 6维评分系统 -sys.path.insert(0, "/home/hmo/MoFin/scripts") -from stock_scorer import score_future_outlook, is_hk_stock, settlement_delay_note - -# ── 趋势检查 ──────────────────────────────────────────────────── -def fetch_trend_data(code): - """取均线数据判断趋势状态。价格从 DB 读取（price_monitor 唯一入口）。返回 (current_price, ma5, trend_label) 或 None""" - # 价格从 DB 读取，不再自拉腾讯 API - current = 0 - try: - db = get_conn() - row = db.execute("SELECT price FROM holdings WHERE code=? AND is_active=1", (code,)).fetchone() - if not row: - row = db.execute("SELECT price FROM watchlist_stocks WHERE code=? AND is_active=1", (code,)).fetchone() - if not row: - row = db.execute("SELECT price FROM holding_strategies WHERE code=? AND status='active' ORDER BY updated_at DESC LIMIT 1", (code,)).fetchone() - if row: - current = row['price'] or 0 - db.close() - except Exception: - pass - - if current <= 0: - return None - - # K线数据仍从腾讯取（均线计算需要历史K线，DB 里 stock_daily 表有但不一定有最新数据） - try: - sym = kline_symbol(code) - if not sym: - return None - url = f"http://ifzq.gtimg.cn/appstock/app/fqkline/get?param={sym},day,,,30,qfq" - req = Request(url, headers={'User-Agent': 'Mozilla/5.0'}) - resp = urlopen(req, timeout=5).read().decode('utf-8') - data = json.loads(resp) - day_key = 'qfqday' if not sym.startswith('hk') else 'day' - bars = data.get('data', {}).get(sym, {}).get(day_key, []) - except: - return None - - if not bars or current <= 0: - return None - closes = [float(b[2]) for b in bars] - if len(closes) < 5: - return None - - def ma(n): - return sum(closes[-n:]) / n - ma5 = ma(5) - ma10 = ma(10) if len(closes) >= 10 else None - ma20 = ma(20) if len(closes) >= 20 else None - - # 趋势分析 - pct_above_ma5 = (current - ma5) / ma5 * 100 - uptrend = False - - if ma20 and ma10: - if ma5 > ma10 > ma20: - trend_label = "多头排列" - uptrend = True - elif current < ma5 and ma5 < ma10 and current < ma10: - trend_label = "空头排列" - elif current > ma5 and ma5 > ma10: - trend_label = "短期转强" - uptrend = True - else: - trend_label = "震荡" - if current > ma5 > ma10: - uptrend = True - else: - trend_label = "数据不足" - - return { - 'price': current, - 'ma5': round(ma5, 2), - 'ma10': round(ma10, 2) if ma10 else None, - 'ma20': round(ma20, 2) if ma20 else None, - 'pct_above_ma5': round(pct_above_ma5, 1), - 'trend': trend_label, - 'uptrend': uptrend, - } - -# ── XMPP -XMPP_BRIDGE = "http://127.0.0.1:5805/" -XMPP_USER = "hmo@yoin.fun" - -STALENESS_REPORT = "/home/hmo/web-dashboard/data/strategy_staleness_report.json" -DETECTOR = "/home/hmo/.hermes/profiles/position-analyst/scripts/stale_detector.py" -REGEN_SCRIPT = "/home/hmo/.hermes/profiles/position-analyst/scripts/per_stock_reassess.py" -REGEN_LOCK = "/tmp/.stale_push_wlin_regen.lock" -MACRO_CTX = "/home/hmo/web-dashboard/data/macro_context.json" -MARKET_JSON = "/home/hmo/web-dashboard/data/market.json" -COOLDOWN_PATH = "/home/hmo/web-dashboard/data/push_cooldown.json" - -NON_BUY_SIGNALS = ["观望", "弱势持有", "深套持有"] - -# 重评冷却：4小时内不重复重评同一股票 -# Dad确认流程: 进区间→重评→(可操作→推荐|不可操作→说明)→冷却期内不再重评+不推重复 -REASSESS_COOLDOWN_HOURS = 4 - - -def get_last_reassess_time(code: str): - """从holding_strategies查最近重评时间""" - try: - db = get_conn() - row = db.execute( - "SELECT updated_at FROM holding_strategies WHERE code=? AND status IN ('active','updated') ORDER BY updated_at DESC LIMIT 1", - (code,) - ).fetchone() - db.close() - if row and row[0]: - return datetime.strptime(row[0][:19], '%Y-%m-%d %H:%M:%S') - except Exception: - pass - return None - - -def is_due_for_reassess(code: str, hours=None) -> bool: - """检查股票是否到重评时间：无历史记录或上次重评超过hours小时""" - if hours is None: - hours = REASSESS_COOLDOWN_HOURS - last = get_last_reassess_time(code) - if last is None: - return True # 从未重评过→需要 - elapsed = datetime.now() - last - return elapsed.total_seconds() > hours * 3600 - - -def load_macro_line(): - """加载大盘和市场的简要描述""" - parts = [] - try: - # 优先 DB - db = get_conn() - row = db.execute( - "SELECT structure FROM macro_context_log " - "WHERE has_valid_data=1 ORDER BY created_at DESC LIMIT 1" - ).fetchone() - db.close() - if row and row[0]: - m = json.loads(row[0]) - else: - raise ValueError("no db data") - overall = m.get("overall", "neutral") - desc = m.get("description", "") - if "bearish" in overall: - parts.append("大盘偏弱") - elif overall == "bullish": - parts.append("大盘偏强") - elif desc: - parts.append(f"大盘{desc}" if not desc.startswith("大盘") else desc) - except Exception: - try: - with open(MACRO_CTX) as f: - m = json.load(f).get("structure", {}) - overall = m.get("overall", "neutral") - desc = m.get("description", "") - if "bearish" in overall: - parts.append("大盘偏弱") - elif overall == "bullish": - parts.append("大盘偏强") - elif desc: - parts.append(f"大盘{desc}" if not desc.startswith("大盘") else desc) - except Exception: - pass - try: - with open(MARKET_JSON) as f: - mk = json.load(f) - mood = mk.get("mood", "") - if mood: - parts.append(f"市场{mood}") - except Exception: - pass - return " | ".join(parts) if parts else "" - - -def is_actionable(cur, timing_signal=""): - """检查信号是否可操作。空文本/含非买入关键词 → 不可操作""" - if not cur and not timing_signal: - return False # 空文本默认不安全 - for kw in NON_BUY_SIGNALS: - if cur and kw.lower() in cur.lower(): - return False - if timing_signal and kw.lower() in timing_signal.lower(): - return False - return True - - -def trigger_regen_sync(stock_codes=None): - """同步执行指定个股的重评（等重评完再发报告）""" - if not stock_codes: - return - try: - cmd = ["python3", REGEN_SCRIPT] + stock_codes - subprocess.run(cmd, capture_output=True, text=True, timeout=60) - except subprocess.TimeoutExpired: - print("[REGEN] 重评超时（60s）", file=sys.stderr) - except Exception as e: - print(f"[REGEN] 重评失败: {e}", file=sys.stderr) - - -def load_cash(): - """从DB实时读可用现金（可用 ≈ 实时买力），不硬编码""" - try: - data = read_portfolio() - if isinstance(data, dict): - # 先读 cash_available（拆分了可用/冻结），fallback 到 cash - return data.get("cash_available", data.get("cash", 0)) - if isinstance(data, list) and len(data) > 1 and isinstance(data[1], dict): - return data[1].get("cash_available", data[1].get("cash", 0)) - return 0 - except Exception: - return 0 - - -_HK_LOT_CACHE = {} - -def hk_lot_size(code): - """从统一入口获取港股实际每手股数，get_price 不提供该字段，默认1000""" - if code in _HK_LOT_CACHE: - return _HK_LOT_CACHE[code] - try: - # 尝试用 get_price 取价，无法获取每手股数，默认1000 - price, chg = get_price(code) - _HK_LOT_CACHE[code] = 1000 - return 1000 - except Exception: - _HK_LOT_CACHE[code] = 1000 - return 1000 - - -def lot_cost(code, price): - if str(code).startswith("688"): - return 200 * price - elif is_hk_stock(code): - lot = hk_lot_size(code) - rate = get_hk_rate() - return int(lot * price * rate) - else: - return 100 * price - - -def push_to_xmpp(text): - """通过知微 HTTP bridge 推送到老爸私信""" - if not text.strip(): - return - try: - payload = json.dumps({ - "to": XMPP_USER, - "body": text.strip(), - "type": "chat", - }).encode("utf-8") - req = Request(XMPP_BRIDGE, data=payload, headers={"Content-Type": "application/json"}) - urlopen(req, timeout=5) - except Exception as e: - print(f"[XMPP推送失败] {e}", file=sys.stderr) - - -def load_cooldown(): - try: - with open(COOLDOWN_PATH) as f: - return json.load(f) - except Exception: - return {} - - -def save_cooldown(cd): - try: - with open(COOLDOWN_PATH, "w") as f: - json.dump(cd, f, indent=2) - except Exception: - pass - - -def in_cooldown(code, action_type, cooldown_dict, minutes=30): - key = f"{code}_{action_type}" - last = cooldown_dict.get(key, 0) - elapsed = datetime.now().timestamp() - last - return elapsed < minutes * 60, elapsed, key - - -def main(): - r = subprocess.run( - ["python3", DETECTOR], capture_output=True, text=True, timeout=60 - ) - if r.returncode != 0 and r.stderr: - print(f"[stderr] {r.stderr.strip()}", file=sys.stderr) - - wl_lines = [ - l for l in r.stdout.split("\n") - if "[WL_IN]" in l and "[自选]" in l - ] - if not wl_lines: - return 0 - - # 读 stale report - try: - with open(STALENESS_REPORT) as f: - report = json.load(f) - except Exception: - report = {"flagged": []} - code_cur = {i["code"]: i.get("current", "") for i in report.get("flagged", [])} - - # 加载冷却状态 - cooldown = load_cooldown() - now_ts = datetime.now().timestamp() - - # ── 从DB读取策略数据 ── - code_data = {} - try: - dec = read_decisions() - for e in dec.get("decisions", []): - code_data[e["code"]] = e - except Exception as _e: - print(f"[DB_LOAD FAIL] {_e}", file=sys.stderr) - - # 2026-08-13 温区自适应：当前温区激活策略集合（regime_weights.active） - _active_strats = None - try: - from regime_gate import _load_weights - _w = _load_weights() - if _w and _w.get("weights"): - _active_strats = {k for k, v in _w["weights"].items() if v.get("matched")} - except Exception: - pass # 无温区数据 → 不过滤（兼容旧逻辑） - - def _is_active(code): - """该自选股的策略是否在当前温区激活""" - if _active_strats is None: - return True - e = code_data.get(code, {}) - for key in ("tag", "version", "strategy_type", "decision_type"): - v = e.get(key) - if v and str(v) in _active_strats: - return True - # 不在激活集（或无标识）→ 视为非激活（保守不推） - return False - - cash = load_cash() - stocks = [] - stale_list = [] - all_candidates = [] # 所有在买入区的自选（stale+non-stale） - - for l in wl_lines: - m = re.match(r'\[WL_IN\](?:\s+\[\w+\])*\s+\[自选\]\s+(\S+)\((\d+)\)', l) - if not m: - continue - name, code = m.group(1), m.group(2) - pm = re.search(r'价(\d+\.\d{2})', l) - if not pm: - continue - price = float(pm.group(1)) - zm = re.search(r'买入([\d.]+)~([\d.]+)', l) - if not zm: - continue - buy_low, buy_high = float(zm.group(1)), float(zm.group(2)) - is_stale = "[STRATEGY_STALE]" in l - cur = code_cur.get(code, "") - - all_candidates.append((name, code, price, buy_low, buy_high, cur, is_stale)) - - if not is_actionable(cur, code_data.get(code, {}).get("timing_signal", "")) or is_stale: - stale_list.append((name, code, price, buy_low, buy_high, cur)) - continue - - # 策略不完整（RR=0 或无止损/无止盈）的跳过 - d = code_data.get(code, {}) - rr = d.get("rr_ratio", 0) or 0 - sl = d.get("stop_loss", 0) or 0 - tp = d.get("take_profit", 0) or 0 - if rr <= 0 or sl <= 0 or tp <= 0: - stale_list.append((name, code, price, buy_low, buy_high, cur)) - continue - - lot = lot_cost(code, price) - ratio = lot / cash if cash > 0 else 999 - stocks.append((name, code, price, buy_low, buy_high, lot, ratio)) - - if not stocks and not stale_list: - return 0 - - now = datetime.now().strftime("%H:%M") - lines = [] - - # 市场背景 - macro_line = load_macro_line() - if macro_line: - lines.append(f"【市场背景】{macro_line}") - - # [关键修复: 2026-07-09] Dad确认流程：进区间→重评→(可操作→推荐|不可操作→说明) - # 冷却期内不再重复重评同一股票：查DB holding_strategies.updated_at - all_codes_in_zone = list(set(s[1] for s in stocks) | set(s[1] for s in stale_list)) - needs_reassess = [c for c in all_codes_in_zone if is_due_for_reassess(c)] - if needs_reassess: - trigger_regen_sync(needs_reassess) - # 重评完成，re-read 最新策略（从DB） - code_data = {} - try: - dec = read_decisions() - for e in dec.get("decisions", []): - code_data[e["code"]] = e - except Exception: - pass - - # 重新过滤：重评后可能有策略变化（止盈/止损/信号变动） - # 去重：同一股票只推送一次（防止两个源重复） - seen_codes = set() - deduped = [] - for item in all_candidates: - code = item[1] - if code not in seen_codes: - seen_codes.add(code) - deduped.append(item) - all_candidates = deduped - - # 补充：从holding_strategies直接扫描可操作的信号（弥补stale_detector遗漏） - try: - import sqlite3 as _sq3 - _db = _sq3.connect('/home/hmo/MoFin/data/mofin.db') - _actionable = _db.execute( - "SELECT hs.code, hs.name, lp.price, lp.change_pct, hs.entry_low, hs.entry_high, " - "hs.stop_loss, hs.take_profit, hs.rr_ratio, hs.timing_signal, hs.action, hs.tech_snapshot, hs.sector_context " - "FROM holding_strategies hs " - "LEFT JOIN live_prices lp ON hs.code = lp.code " - "WHERE hs.status='active' AND hs.timing_signal IN ('买入','可买入','可加仓') " - "AND hs.rr_ratio > 0 AND hs.stop_loss > 0 AND hs.take_profit > 0" - ).fetchall() - _db.close() - for row in _actionable: - code = str(row[0]) - if code in seen_codes: - continue - price = row[2] or 0 - el = row[4] or 0 - eh = row[5] or 0 - if price <= 0 or el <= 0 or eh <= 0: - continue - # 价格必须在买入区内（硬检查，拒绝偏离仍推买入） - if price < el or price > eh: - continue - name = row[1] or code - all_candidates.append((name, code, price, el, eh, "", False)) - seen_codes.add(code) - except Exception: - pass - - # 重建 stocks 列表，用新数据判断（不再用旧 is_stale 标记，因为已全部重评） - stocks = [] - zone_notes = [] # 在操作区间但不可操作→发说明 - for (name, code, price, buy_low, buy_high, cur, is_stale) in all_candidates: - sig = code_data.get(code, {}).get("timing_signal", "") - d = code_data.get(code, {}) - rr = d.get("rr_ratio", 0) or 0 - sl = d.get("stop_loss", 0) or 0 - tp = d.get("take_profit", 0) or 0 - - # 判断重评后的可操作性 - reason = "" - if not is_actionable(cur, sig): - reason = f"信号'{sig}'非可操作方向" - elif rr <= 0 or sl <= 0 or tp <= 0: - reason = f"策略不完整(RR={rr} 损={sl} 盈={tp})" - elif any(kw in sig for kw in ["等企稳", "信号不充分"]): - reason = f"信号'{sig}'，暂不建议操作" - - if reason: - # 冷却检查：同股同原因4小时内不发（匹配重评冷却） - ck = f"zone_note_{code}" - now_ts = datetime.now().timestamp() - last = cooldown.get(ck, 0) - if now_ts - last > REASSESS_COOLDOWN_HOURS * 3600: - zone_notes.append((name, code, price, buy_low, buy_high, reason, sig)) - cooldown[ck] = now_ts - continue - - lot = lot_cost(code, price) - ratio = lot / cash if cash > 0 else 999 - stocks.append((name, code, price, buy_low, buy_high, lot, ratio)) - - # 加载portfolio获取持仓信息（A/H去重用） - pf = {"holdings": []} - try: - pf = read_portfolio() - except Exception: - pass - - stocks.sort(key=lambda s: ( - 0 if len(str(s[1])) == 6 else 1, - -code_data.get(s[1], {}).get("rr_ratio", 0) - )) - - # 只展示有清晰操作信号的个股 - # timing_signal 必须是明确操作方向：买入/加仓/观望/关注/信号不充分 - # 行业描述（行业偏弱/行业偏强/大盘变盘等）不是操作信号，一律跳过 - VALID_SIGNALS = {"买入", "加仓", "观望", "关注", "信号不充分"} - SKIP_KEYWORDS = ["等企稳", "信号不充分"] - - actionable = [] - for s in stocks: - sig = code_data.get(s[1], {}).get("timing_signal", "") - if not sig: - continue - # 跳过非操作信号 - if any(kw in sig for kw in SKIP_KEYWORDS): - continue - # 中性信号跳过 - stripped = sig.strip() - if not stripped or stripped.lower() in ("", "neutral", "持有", "深套持有", "弱势持有"): - continue - # 信号必须含买入/加仓才推荐——其他非操作信号跳过 - if not any(kw in sig for kw in ["买入", "加仓"]): - continue - # 价格必须在买入区内（硬检查，拒绝价格偏离仍推买入） - buy_low, buy_high = s[3], s[4] - price_check = s[2] - if buy_low and buy_high and buy_low > 0: - if not (buy_low <= price_check <= buy_high): - continue - # RR完整性检查：买入/加仓信号必须RR>0（策略数据要完整） - cd = code_data.get(s[1], {}) - rr = cd.get("rr_ratio", 0) or 0 - tp = cd.get("take_profit", 0) or 0 - if rr <= 0 or tp <= 0: - # 策略数据不完整（缺止盈/RR），不推 - continue - # 趋势检查：必须不是空头排列（价格在MA5以下且MA5 cash: - continue - - actionable.append(s) - - if not actionable and not zone_notes: - return 0 # 无推荐也无区间说明 → 静默 - - # 加载基本面缓存（PE等） - fund_cache = {} - try: - import multi_timeframe as mtf_mod - mtf = mtf_mod._load_mtf_cache() - for code, v in mtf.items(): - fund_cache[code] = v.get("fundamentals", {}) - except Exception: - pass - - # 仓位计算：从DB读取总资产和现金 - n = len(actionable) - total_assets = 0 - available_cash = 0 - try: - pf = read_portfolio() - available_cash = pf.get("cash_available", pf.get("cash", 0)) or 0 - # 直接取 portfolio 的总资产（导入时已做港币→人民币换算） - total_assets = pf.get("total_assets", 0) or 0 - if total_assets <= 0: - # fallback: use unified calc_total_assets from mo_models - total_assets = calc_total_assets(pf) - except Exception: - total_assets = available_cash * 5 # fallback - - # 加载策略树模块（获取当前情景+分支评估） - st = None - scenario_id = "" - scenario_label = "" - try: - import importlib.util - spec = importlib.util.spec_from_file_location("st_module", "/home/hmo/MoFin/strategy_tree.py") - st = importlib.util.module_from_spec(spec) - spec.loader.exec_module(st) - sc = st.detect_scenario() - scenario_id = sc.get("id", "") - scenario_label = sc.get("label", "") - except Exception: - pass - - def calc_position(lot_cost, rr, market_factor, cat, code=""): - # 理论推荐仓位(% of 总资产) — 仅基于RR+市场+品种，不受现金限制 - if rr >= 5: - theo_pct = 25 - elif rr >= 3: - theo_pct = 18 - elif rr >= 2: - theo_pct = 12 - else: - theo_pct = 8 - if "偏弱" in market_factor: - theo_pct = int(theo_pct * 0.8) - elif "偏强" in market_factor: - theo_pct = int(theo_pct * 1.15) - if cat in ("蓝筹", "白马"): - theo_pct = int(theo_pct * 1.2) - elif cat in ("题材", "短线"): - theo_pct = int(theo_pct * 0.6) - elif cat in ("高波动", "成长"): - theo_pct = int(theo_pct * 0.85) - theo_pct = max(5, min(30, theo_pct)) - - # 当前建议仓位：理论占总资产% → 按现金锁死 - ideal_budget = total_assets * theo_pct / 100 - # 可操作N只时，现金分配不超过 available_cash / n * 1.5 - max_use_cash = (available_cash / max(n, 1)) * 1.5 - budget = min(ideal_budget, max_use_cash, available_cash) - lots = int(budget / lot_cost) if lot_cost > 0 else 0 - - if lots == 0 and lot_cost > 0 and budget > lot_cost * 0.8: - # 预算覆盖超过80%的1手金额 → 至少1手（仅差一档） - lots = 1 - - lot_cost_total = lots * lot_cost - if lots == 0: - pct_actual = 0 - elif total_assets > 0: - pct_actual = round(lot_cost_total / total_assets * 100) - else: - pct_actual = 0 - - if lots == 0: - details = f"预算不足1手（{budget:,.0f}/{lot_cost:,.0f}元）" - else: - if len(str(code)) == 5: - hk_lot = hk_lot_size(code) - shares = lots * hk_lot - elif code.startswith("688"): - shares = lots * 200 - else: - shares = lots * 100 - details = f"{lots}手（{shares}股，{lot_cost_total:,.0f}元）" - - return theo_pct, pct_actual, details, lots, lot_cost_total - - # ── 换仓评估 ────────────────────────────────────────────────────── - # score_future_outlook 从 stock_scorer 模块导入（6维评分） - - def evaluate_swap(lot_cost_target, rr, sig, tp, sl, name, code, price_in, - total_assets_in, cash_in, pf_in, cd_in): - """现金不足时评估是否卖差票换推荐股。 - - 核心逻辑： - - 已发生的亏损是沉没成本，不参与决策 - - 用6维评分法评估每个持仓的未来前景（基于决策系统既有数据） - - 优先卖前景最差的票，保留前景好的票（无论当前盈亏%） - - 卖港股→买A股需T+2到账，如果推荐此方案则标注延迟风险 - - 对目标票(RR>=3+买入信号)才有换仓资格 - - 返回（推荐文案str, 缺口float）或 (None, gap) - """ - gap = lot_cost_target - cash_in - # 目标票质量门槛 - if rr < 3.0 or gap <= 0 or gap > total_assets_in * 0.5: - return None, gap - if not any(kw in sig for kw in ["买入", "加仓", "建仓"]): - return None, gap - - # 收集持仓数据 + 前景评分 - ph = [] - for h in pf_in.get("holdings", []): - hs = h.get("shares", 0) or 0 - hp = h.get("price", 0) or 0 - hc = h.get("cost", 0) or 0 - if hs <= 0 or hp <= 0: - continue - hmv = hs * hp - # 港股价格已是 CNY（price_monitor 写入时已转），不需要再乘汇率 - hpl_pct = (hp - hc) / hc * 100 if hc else 0 - - # 6维全面评分（越低越差，越建议卖） - fscore, _ = score_future_outlook(h_code, cd_in) - - ph.append({ - "code": h_code, - "name": h.get("name", ""), - "shares": hs, - "price": hp, - "cost": hc, - "mv": round(hmv), - "pl_pct": round(hpl_pct, 1), - "score": fscore, - }) - - # 按前景评分升序（最差的排最前面） - ph.sort(key=lambda x: x["score"]) - - # 打印调试信息：所有持仓的前景评分 - # print(f"[SWAP_DEBUG] 前景评分(越低越差):", file=sys.stderr) - # for x in ph[:10]: - # print(f" {x['name']}({x['code']}) 评分{x['score']} 亏{x['pl_pct']}% 市值{x['mv']:,}", file=sys.stderr) - - # 只考虑评分<=0（前景差或中性偏弱）的作为减仓候选 - candidates = [h for h in ph if h["score"] <= 0] - if not candidates: - return None, gap - - # 贪心选评分最差的，凑够现金缺口（最多2只） - selected = [] - cash_freed = 0 - for h in candidates: - if cash_freed >= gap: - break - cash_freed += h["mv"] - selected.append(h) - - if cash_freed < gap or len(selected) > 2: - return None, gap - - # 计算目标票的预期涨幅 - if tp and tp > 0: - target_gain_pct = (tp - price_in) / price_in * 100 - else: - target_gain_pct = rr * 3 - - # 构建推荐文案 - buy_is_a = not is_hk_stock(code) # 目标是否是A股 - sell_parts = [] - sell_names = [] - settlement_warnings = [] - for h in selected: - # 每个被选股票配一句"为什么卖它" - reason = f"评分{h['score']}" - if h['pl_pct'] <= -30: - reason += "深套" - elif h['pl_pct'] <= -15: - reason += f"亏损{h['pl_pct']}%" - sell_parts.append(f"{h['name']}({h['code']}) {h['shares']}股 亏{h['pl_pct']}% ({reason})") - sell_names.append(h['name']) - # 检查结算延迟：卖港股→买A股 - if is_hk_stock(h['code']) and buy_is_a: - settlement_warnings.append(f"{h['name']}是港股通，卖出需T+2到账才能买A股") - sell_desc = "；".join(sell_parts) - - new_budget = cash_in + cash_freed - new_lots = int(new_budget / lot_cost_target) if lot_cost_target > 0 else 0 - if new_lots == 0: - return None, gap - if code.startswith("688"): - new_shares = new_lots * 200 - elif len(code) <= 5: - new_shares = new_lots * hk_lot_size(code) - else: - new_shares = new_lots * 100 - new_cost = new_lots * lot_cost_target - new_pct = round(new_cost / total_assets_in * 100) if total_assets_in > 0 else 0 - - text = ( - f"换仓建议：卖{sell_desc}" - f"→腾{round(cash_freed):,}元" - f"→买{name}({code}) {new_lots}手({new_shares}股,{round(new_cost):,}元)" - f"占{new_pct}%仓位" - f"（止损{sl}(-{round((price_in-sl)/price_in*100,1)}%)" - f"止盈{tp}(+{round(target_gain_pct,1)}%)" - f" RR={rr}）\n" - f" 理由：{', '.join(sell_names)}评分最低，" - f"继续持有无积极信号且技术偏弱；" - f"换到有明确信号和止损的标的，预期收益更优。" - ) - if settlement_warnings: - text += "\n ⚠️ " + " | ".join(settlement_warnings) - return text, gap - - # 标准格式：每个可操作标的 — 大盘/行业/个股三面 + 仓位 - if actionable: - lines.append(f"【💡 操作建议】（当前{len(actionable)}只自选可操作 | 总资产{total_assets:,.0f}元 现金{available_cash:,.0f}元）") - for s in actionable: - name, code, price, buy_low, buy_high, lot, ratio = s - d = code_data.get(code, {}) - sl = d.get("stop_loss", 0) - tp = d.get("take_profit", 0) - rr = d.get("rr_ratio", 0) - sig = d.get("timing_signal", "") - sector = d.get("sector_context", "") - tech = d.get("tech_snapshot", "") - mtf_ctx = d.get("multi_tf_context", "") - note = d.get("note", "") - d_factors = d.get("signal_factors", []) - cat = d.get("stock_category", "") - - # 提取技术位 - ss = {"强撑":"-", "弱撑":"-", "弱压":"-", "强压":"-"} - for tag in ss: - m = re.search(rf'{tag}:([\d.]+)', tech) - if m: - ss[tag] = m.group(1) - - # 基本面 - fund = fund_cache.get(code, {}) - pe = fund.get("pe", 0) - eps = fund.get("eps", 0) - pe_str = f"PE{pe:.0f}" if pe else "" - eps_str = f"EPS{eps:.2f}" if eps else "" - - # 从 signal_factors 提取各维度 - def _match_factor(prefix): - for f in d_factors: - if f.startswith(prefix): - return f - return "" - - market_factor = _match_factor("大盘") - sector_factor = _match_factor("行业") - value_factor = _match_factor("高估值") or _match_factor("低估值") or _match_factor("蓝筹") or pe_str or "" - news_factor = _match_factor("消息") - tech_factor = _match_factor("净利") or _match_factor("组合") or "" - - # 构建分析行 - parts = [] - if market_factor: - parts.append(f"大盘{market_factor.replace('大盘','')}") - if sector_factor: - parts.append(f"行业{sector_factor.replace('行业','')}") - if pe_str or value_factor: - parts.append(value_factor or pe_str) - if news_factor: - parts.append(news_factor) - if not parts: - parts.append(sector or cat or "") - - analysis = " | ".join(p for p in parts if p) - - # 仓位计算 - theo_pct, actual_pct, details, lots, lot_cost_total = calc_position( - lot, rr, market_factor, cat, code - ) - - pfx = "" if len(code) == 6 else "HK$" - - # 取分支动作类型 - branch_action = "hold" - branch_rationale = "" - if st and scenario_id: - try: - results = st.evaluate_branches(code, scenario_id, price, d.get("shares", 0), d.get("cost", 0)) - applicable = [r for r in results if r.get("applicable")] - if applicable: - best = min(applicable, key=lambda r: r.get("priority", 999)) - branch_action = best.get("action_type", "hold") - branch_rationale = best.get("rationale", "") - except Exception: - pass - - # 冷却检查：相同股+相同操作30分钟内不发 - cooled, elapsed, cd_key = in_cooldown(code, branch_action, cooldown) - if cooled: - continue - - # 策略质量过滤：只有正向/中性信号才推操作建议 - bad_keywords = ["偏弱", "弱势", "观望", "卖出", "回避", "回避"] - if any(kw in sig for kw in bad_keywords): - continue - - # 行业背景过滤：行业大跌时不在买入区推荐（即使个股信号好） - if "大跌" in sector: - continue - - # 换仓评估：现金不足时评估是否卖差票换推荐股 - swap_text = None - if lots == 0: - swap_text, _ = evaluate_swap( - lot, rr, sig, tp, sl, name, code, price, - total_assets, available_cash, pf, code_data - ) - - action_tag = "🛒" if (lots > 0 or swap_text) else "⚠️" - - # 2026-08-13 信号溯源：记录策略/版本/温区 + 共振标记（多策略同推加关注） - strat_id = d.get("tag") or d.get("version") or d.get("strategy_type") or d.get("decision_type") or "" - regime_now = "" - temp_now = "" - try: - # 2026-08-14 阶段4：温区溯源按标的市场（A股=原行为, 港股=港股温区） - from market_config import get_regime_temp - regime_now, temp_now = get_regime_temp(code) - except Exception: - pass - # 溯源记录 + 共振检测 - resonance_tag = "" - try: - from signal_ledger import record_signal - _res = record_signal( - code=code, name=name, strategy=strat_id, version=strat_id, - regime=regime_now, temp_band=temp_now, - reason=f"进买入区{buy_low}~{buy_high} + 重评{sig}", - source_module="stale_push_wlin", - ) - if _res and _res.get("resonance_count", 1) > 1: - resonance_tag = f" 🔥多策略共振({_res['resonance_count']}策略: {_res['resonance_strategies']})" - except Exception: - pass - strat_tag = f" [{strat_id}]" if strat_id else "" - regime_tag = f"[{regime_now}]" if regime_now else "" - - lines.append( - f" {action_tag} {name}({code}) {pfx}{price:.2f} 买区{buy_low}~{buy_high} | " - f"1手{lot:,.0f}元 RR={rr:.1f} 损{sl} 盈{tp}{strat_tag}{regime_tag}{resonance_tag}\n" - f" {analysis}\n" - f" 技术{ss['强撑']}→{ss['弱撑']}→{ss['弱压']}→{ss['强压']} | 信号{sig}\n" - f" 仓位：理论{theo_pct}%×总资产 | 建议{actual_pct}%（{details}）" - ) - - if mtf_ctx: - lines[-1] += f"\n 均线{mtf_ctx}" - - if swap_text: - lines[-1] += f"\n {swap_text}" - - # 分支描述 - branch_line = "" - if branch_action != "hold": - branch_line = f" 【{scenario_label}→{branch_action}】{branch_rationale}" - if branch_line: - lines[-1] += f"\n{branch_line}" - - # 记录推送时间（冷却计时用） - cooldown[cd_key] = now_ts - - save_cooldown(cooldown) - - # 修正可操作数量（剔除冷却跳过后的实际数量） - if actionable: - actual_n = sum( - 1 for ln in lines - if ln.startswith(" 🛒") or ln.startswith(" ⚠️") - ) - if actual_n != len(actionable): - for i, ln in enumerate(lines): - if "【💡 操作建议】" in ln: - if actual_n > 0: - lines[i] = f"【💡 操作建议】（当前{actual_n}只自选可操作 | 总资产{total_assets:,.0f}元 现金{available_cash:,.0f}元）" - else: - lines.pop(i) # 全部冷却，移除空标题 - break - - # 检查最终是否还有内容要推 - has_actionable = any( - ln.startswith(" 🛒") or ln.startswith(" ⚠️") for ln in lines - ) - if not has_actionable: - return 0 # 2026-08-19 落实老莫8/14决定：无操作直接静默（zone_notes只记日志不推送） - - # ── T+2前瞻：扫描近期可能入买区的A股，提前准备现金 ── - t2_lines = [] - try: - dec_t2 = read_decisions() - for entry in dec_t2.get("decisions", []): - if entry.get("status") == "closed" or entry.get("type") != "自选策略": - continue - ec = entry["code"] - el = entry.get("entry_low", 0) or 0 - eh = entry.get("entry_high", 0) or 0 - ep = entry.get("price", 0) or 0 - if not eh or not ep or el <= 0: - continue - # A股+价格在买入区上方5%以内（即将进入买入区） - if not is_hk_stock(ec) and el <= ep <= eh * 1.05 and ep > eh: - anticipation_pct = (ep - eh) / eh * 100 - lot = lot_cost(ec, ep) - if lot > available_cash: - # 现金不足 → 卖港股提前准备 - ph = [] - for h in pf.get("holdings", []): - hs = h.get("shares", 0) or 0 - hp = h.get("price", 0) or 0 - hc = h.get("cost", 0) or 0 - if hs <= 0 or hp <= 0 or not is_hk_stock(h.get("code","")): - continue - sc = score_future_outlook(h.get("code",""), code_data) - ph.append((sc, h)) - ph.sort(key=lambda x: x[0]) - if ph: - worst = ph[0][1] - w_name = worst.get("name","?") - w_code = worst.get("code","") - w_price = worst.get("price",0) - w_shares = worst.get("shares",0) - w_value = w_price * w_shares - if w_value >= lot: - name_e = entry.get("name","") - t2_lines.append( - f" ⏳ {name_e}({ec})距买入区仅{anticipation_pct:.0f}%，" - f"需{lot:,.0f}元。建议提前卖{w_name}({w_code})" - f"腾{w_value:,.0f}元(T+2到账后可用)" - ) - except: - pass - - if t2_lines: - lines.append("") - lines.append("【⏳ 提前准备（T+2港股提前出清）】") - lines.extend(t2_lines) - - # 2026-08-14 删除：操作区间内但重评后不可操作的推送（垃圾消息，不推荐不推） - # zone_notes 仅记日志，不推送（老莫：进入区间但重评不推荐=垃圾消息） - if zone_notes: - print(f"[LOG] {len(zone_notes)} 只进入区间但重评不推荐（不推送）: {[z[0] for z in zone_notes[:5]]}", flush=True) - - # 标题：有推荐操作→"自选买入提醒"，仅有区间说明→"操作区间提醒" - if has_actionable: - lines.insert(0, f"【知微】自选买入提醒 {now} | 总资产{total_assets:,.0f}元") - else: - lines.insert(0, f"【知微】操作区间提醒 {now} | 总资产{total_assets:,.0f}元") - out = "\n".join(lines) - print(out) - push_to_xmpp(out) - return 0 - - -if __name__ == "__main__": - sys.exit(main())