feat: 统一部署目录——所有运行时文件归入MoFin repo

- deploy/bot/ — XMPP bot核心(xmpp_agent_core + xmpp_zhiwei_bot) - deploy/profile-scripts/ — cron脚本(price_monitor等) - 运行时文件已替换为指向MoFin的符号链接 - 改代码只需改MoFin，系统自动生效
2026-07-17 23:12:35 +08:00
parent ab74188915
commit 80d59c9331
103 changed files with 29293 additions and 0 deletions
@@ -0,0 +1,742 @@
+#!/usr/bin/env python3
+"""price_monitor.py — 高频价格监控脚本（批量版）
+规则：进入区间报一次，离开区间报一次，中间不重复。
+每次运行时一次性刷新所有持仓+自选股的实时价。
+"""
+import urllib.request
+import os, sys, time, json
+import sqlite3
+from datetime import datetime
+
+from mo_data import read_decisions
+
+BREACH_PATH = "/home/hmo/.hermes/zone_breach.json"
+STATE_PATH = "/home/hmo/.hermes/price_trigger_state.json"
+EVENTS_PATH = "/home/hmo/web-dashboard/data/price_events.json"
+
+# DB 模块（同步实时价到 mofin.db）
+sys.path.insert(0, "/home/hmo/MoFin")
+try:
+    from mofin_db import get_conn, DB_PATH
+    from mo_models import calc_total_mv, calc_total_assets
+    HAS_DB = True
+except ImportError:
+    HAS_DB = False
+
+# 策略重评依赖（技术面驱动，非机械百分比）
+sys.path.insert(0, "/home/hmo/web-dashboard")
+try:
+    from strategy_lifecycle import reassess_strategy, reassess_with_context
+    HAS_REASSESS = True
+except ImportError:
+    HAS_REASSESS = False
+
+UA = "Mozilla/5.0"
+
+# ── XMPP推送 ──────────────────────────────────────────────────────────
+XMPP_USER = "hmo@yoin.fun"
+XMPP_BRIDGE = "http://127.0.0.1:5805/"
+
+def push_to_xmpp(text):
+    """通过知微 HTTP bridge 推送到Dad私信"""
+    if not text.strip():
+        return
+    try:
+        payload = json.dumps({
+            "to": XMPP_USER,
+            "body": text.strip(),
+            "type": "chat",
+        }).encode("utf-8")
+        req = urllib.request.Request(XMPP_BRIDGE, data=payload, headers={"Content-Type": "application/json"})
+        urllib.request.urlopen(req, timeout=5)
+    except Exception as e:
+        print(f"[XMPP推送失败] {e}", file=sys.stderr)
+
+# ── 批量拉取价格 ──────────────────────────────────────────────────────────
+
+def fetch_all_prices(codes):
+    """腾讯批量行情API：一次请求拉取所有股票（A股+港股）
+    A股：sh600110 / sz000001
+    港股：hk00700
+    返回 {code: (price, change, change_pct)}
+    """
+    if not codes:
+        return {}
+
+    # 构建批量查询串
+    symbols = []
+    code_map = {}  # symbol -> original_code
+    for code in codes:
+        code_s = str(code).strip()
+        if len(code_s) == 6:
+            # A股：沪市以5/6/9开头，深市以0/3开头
+            if code_s.startswith(('5', '6', '9')):
+                sym = f"sh{code_s}"
+            else:
+                sym = f"sz{code_s}"
+        else:
+            sym = f"hk{code_s}"
+        symbols.append(sym)
+        code_map[sym] = code_s
+
+    url = f"http://qt.gtimg.cn/q={','.join(symbols)}"
+    try:
+        req = urllib.request.Request(url, headers={"User-Agent": UA})
+        with urllib.request.urlopen(req, timeout=10) as r:
+            text = r.read().decode("gbk")
+    except Exception as e:
+        print(f"⚠️ 批量拉取失败: {e}", file=sys.stderr)
+        return {}
+
+    results = {}
+    for line in text.strip().split("\n"):
+        line = line.strip()
+        if not line or "=" not in line:
+            continue
+        try:
+            # 格式: v_sh600110="1~诺德股份~600110~11.84~11.90~..."
+            raw_value = line.split("=", 1)[1].strip().strip('"').strip(";")
+            fields = raw_value.split("~")
+            if len(fields) < 6:
+                continue
+            sym = line.split("=", 1)[0].strip().lstrip("v_")
+            orig_code = code_map.get(sym)
+            if not orig_code:
+                continue
+            price = float(fields[3]) if fields[3] else 0
+            prev_close = float(fields[4]) if fields[4] else 0
+            change = price - prev_close if prev_close > 0 else 0
+            change_pct = fields[32] if len(fields) > 32 and fields[32] else "0"
+            results[orig_code] = (price, change, change_pct)
+        except (ValueError, IndexError):
+            continue
+
+    return results
+
+
+def refresh_data_prices():
+    """一次性刷新所有持仓+自选股的实时价（完全DB版，不写JSON）"""
+    all_codes = set()
+
+    # 从DB读所有需要拉取价格的代码
+    try:
+        conn = get_conn()
+        for r in conn.execute("SELECT code FROM holdings WHERE is_active=1"):
+            all_codes.add(r['code'])
+        for r in conn.execute("SELECT code FROM watchlist_stocks"):
+            all_codes.add(r['code'])
+        for r in conn.execute("SELECT code FROM holding_strategies WHERE status='active'"):
+            all_codes.add(r['code'])
+        conn.close()
+    except Exception as e:
+        print(f"⚠️ 从DB读代码失败: {e}", file=sys.stderr)
+        return 0
+
+    if not all_codes:
+        return 0
+
+    # 一次性批量拉取
+    prices = fetch_all_prices(list(all_codes))
+    updated = len(prices)
+
+    # === 弹性同步实时价到 mofin.db ===
+    # 防死锁策略（经2026-07-14 WAL死锁复盘改进）：
+    #  ① 启动时 checkpoint WAL（清理残留事务）
+    #  ② 统一 BEGIN IMMEDIATE 包裹整个写操作
+    #  ③ 5次重试 + 指数退避: 1s → 2s → 4s → 8s → 16s（共~31s）
+    #  ④ get_conn() 的 busy_timeout=30000 保证等待上限
+    #  ⑤ 每个写操作检查返回值，任一失败立即 rollback + 重试
+    #  ⑥ try/finally 确保连接始终释放
+    if HAS_DB and prices:
+        # 先checkpoint一次，清理上次被kill残留的WAL
+        try:
+            c = get_conn()
+            c.execute("PRAGMA wal_checkpoint(TRUNCATE)")
+            c.close()
+        except Exception:
+            pass
+
+        max_tries = 5
+        conn = None
+        for db_attempt in range(max_tries):
+            try:
+                conn = get_conn()
+                # BEGIN IMMEDIATE 立即获取写锁——失败则等 busy_timeout(30s)
+                conn.execute("BEGIN IMMEDIATE")
+
+                # ── 构建 holdings 更新数据 ──
+                db_holdings = []
+                for r in conn.execute("SELECT * FROM holdings WHERE is_active=1"):
+                    h = dict(r)
+                    code = str(h.get('code', ''))
+                    if code in prices:
+                        price_val, _, change_pct = prices[code]
+                        if price_val > 0:
+                            h['price'] = round(price_val, 2)
+                            h['change_pct'] = float(change_pct) if change_pct else 0
+                    db_holdings.append(h)
+
+                # ── 写 holdings 表 ──
+                for h in db_holdings:
+                    currency = str(h.get('currency', 'CNY')).upper()
+                    if currency not in ('CNY', 'HKD'):
+                        raise ValueError(f"非法币种: {currency}")
+                    conn.execute("""
+                        INSERT INTO holdings (code, name, shares, cost, price, market_value,
+                            change_pct, currency, position_pct, added_at, is_active)
+                        VALUES (?,?,?,?,?,?,?,?,?,datetime('now','localtime'),1)
+                        ON CONFLICT(code) DO UPDATE SET
+                            name=excluded.name, shares=excluded.shares, cost=excluded.cost,
+                            price=excluded.price, market_value=excluded.market_value,
+                            change_pct=excluded.change_pct, currency=excluded.currency,
+                            position_pct=excluded.position_pct
+                    """, (
+                        h.get('code'), h.get('name'), h.get('shares', 0),
+                        h.get('cost'), h.get('price'),
+                        h.get('market_value'), h.get('change_pct'),
+                        h.get('currency', 'CNY'), h.get('position_pct'),
+                    ))
+
+                # ── 写 portfolio_summary ──
+                mv = calc_total_mv(db_holdings)
+                existing = conn.execute(
+                    'SELECT cash, frozen_cash FROM portfolio_summary WHERE id=1'
+                ).fetchone()
+                db_cash = existing['cash'] if existing else 0.0
+                db_frozen = existing['frozen_cash'] if existing else 0.0
+                assets = calc_total_assets({'holdings': db_holdings, 'cash': db_cash, 'frozen_cash': db_frozen})
+                position_pct = round(mv / assets * 100, 2) if assets > 0 else 0
+                conn.execute("""
+                    INSERT INTO portfolio_summary (id, total_assets, total_mv, stock_value,
+                        cash, frozen_cash, position_pct, total_pnl, currency, updated_at)
+                    VALUES (1,?,?,?,?,?,?,?,?,datetime('now','localtime'))
+                    ON CONFLICT(id) DO UPDATE SET
+                        total_assets=excluded.total_assets, total_mv=excluded.total_mv,
+                        stock_value=excluded.stock_value, cash=excluded.cash,
+                        frozen_cash=excluded.frozen_cash, position_pct=excluded.position_pct,
+                        total_pnl=excluded.total_pnl, currency=excluded.currency,
+                        updated_at=datetime('now','localtime')
+                """, (
+                    assets, mv, mv, db_cash, db_frozen,
+                    position_pct, 0, 'CNY',
+                ))
+
+                # ── 写 live_prices ──
+                for h in db_holdings:
+                    code = h.get('code', '')
+                    if code:
+                        p = h.get('price', 0)
+                        cp = h.get('change_pct', 0)
+                        conn.execute(
+                            "INSERT OR REPLACE INTO live_prices (code, price, change_pct, updated_at) "
+                            "VALUES (?,?,?,datetime('now','localtime'))",
+                            (code, p, cp)
+                        )
+                # 补充策略股/自选股的价格（不在holdings中的）
+                for code, pdata in prices.items():
+                    if code not in {h.get('code') for h in db_holdings}:
+                        price_val = pdata[0] if isinstance(pdata, (list, tuple)) else pdata.get('price', 0)
+                        cp_val = pdata[1] if isinstance(pdata, (list, tuple)) else pdata.get('change_pct', 0)
+                        conn.execute(
+                            "INSERT OR REPLACE INTO live_prices (code, price, change_pct, updated_at) "
+                            "VALUES (?,?,?,datetime('now','localtime'))",
+                            (code, price_val, cp_val)
+                        )
+
+                conn.commit()
+                conn.close()
+                conn = None
+                if db_attempt > 0:
+                    print(f"DB同步成功（第{db_attempt+1}次重试）")
+                break  # success
+
+            except (sqlite3.OperationalError, sqlite3.DatabaseError) as e:
+                if conn:
+                    try: conn.rollback()
+                    except Exception: pass
+                    try: conn.close()
+                    except Exception: pass
+                    conn = None
+                err_str = str(e)
+                if "locked" in err_str or "cannot commit" in err_str or "busy" in err_str:
+                    if db_attempt < max_tries - 1:
+                        wait = 2 ** db_attempt  # 1, 2, 4, 8, 16
+                        print(f"⏳ DB锁(尝试{db_attempt+1}/{max_tries}): {e} → {wait}s后重试", file=sys.stderr)
+                        time.sleep(wait)
+                    else:
+                        print(f"❌ DB锁(重试{max_tries}次耗尽): {e}", file=sys.stderr)
+                else:
+                    print(f"❌ DB错误: {e}", file=sys.stderr)
+                    break
+            except Exception as e:
+                if conn:
+                    try: conn.rollback()
+                    except Exception: pass
+                    try: conn.close()
+                    except Exception: pass
+                    conn = None
+                print(f"⚠️ DB同步异常: {e}", file=sys.stderr)
+                break
+        else:
+            # for-else: loop exhausted without break
+            print("❌ DB同步失败（所有重试耗尽）", file=sys.stderr)
+            # 尝试紧急 WAL checkpoint（释放死锁）
+            try:
+                c = sqlite3.connect(str(DB_PATH), timeout=1)
+                c.execute("PRAGMA wal_checkpoint(TRUNCATE)")
+                c.close()
+                print("  ↪ 紧急WAL checkpoint完成", file=sys.stderr)
+            except Exception as we:
+                print(f"  ↪ WAL checkpoint也失败: {we}", file=sys.stderr)
+
+    return updated
+
+
+# ── 区间偏离检测 ──────────────────────────────────────────────────────────
+
+def load_state():
+    try:
+        with open(STATE_PATH) as f:
+            return json.load(f)
+    except:
+        return {}
+
+def save_state(state):
+    os.makedirs(os.path.dirname(STATE_PATH), exist_ok=True)
+    with open(STATE_PATH, 'w') as f:
+        json.dump(state, f, ensure_ascii=False, indent=2)
+
+def load_breaches():
+    try:
+        with open(BREACH_PATH) as f:
+            return json.load(f)
+    except:
+        return {}
+
+def save_breaches(data):
+    os.makedirs(os.path.dirname(BREACH_PATH), exist_ok=True)
+    with open(BREACH_PATH, 'w') as f:
+        json.dump(data, f, ensure_ascii=False, indent=2)
+
+
+def load_events():
+    try:
+        with open(EVENTS_PATH) as f:
+            return json.load(f)
+    except:
+        return {"events": []}
+
+
+def save_events(events):
+    os.makedirs(os.path.dirname(EVENTS_PATH), exist_ok=True)
+    with open(EVENTS_PATH, 'w') as f:
+        json.dump(events, f, ensure_ascii=False, indent=2)
+
+
+def record_event(code, name, event_type, price, trigger_value, event_label=""):
+    """记录一次价格触发事件到 price_events.json"""
+    events = load_events()
+    now = datetime.now().isoformat()
+    events["events"].append({
+        "code": code,
+        "name": name,
+        "event_type": event_type,  # entry_zone, stop_loss, take_profit, exit_zone
+        "price": round(price, 2),
+        "trigger_value": trigger_value,
+        "event_label": event_label,
+        "timestamp": now,
+        "date": datetime.now().strftime("%Y-%m-%d"),
+    })
+    # 保留最近10000条
+    events["events"] = events["events"][-10000:]
+    save_events(events)
+
+
+def get_trigger_zones(trigger):
+    """返回该trigger所有可监控的区间列表，跳过已执行的batch"""
+    zones = []
+    for key, label in [
+        ("entry_zone", "加仓区间"),
+        ("batch1_price", "试仓区间"),
+        ("batch2_price", "加仓区间"),
+        ("take_profit_zone", "止盈区间"),
+        ("watch_low", "关注区间"),
+        ("watch_high", "减仓区间"),
+        ("watch_break", "止损区间")
+    ]:
+        status_key = key.replace("_price", "_status")
+        if status_key in trigger and trigger[status_key] == "executed":
+            continue
+        val = trigger.get(key, "")
+        if val and "~" in val:
+            try:
+                parts = val.split("~")
+                lo, hi = float(parts[0]), float(parts[1])
+                zones.append((key, label, lo, hi))
+            except:
+                pass
+    sl = trigger.get("stop_loss", "")
+    if sl:
+        try:
+            sl_price = float(sl) if isinstance(sl, (int, float)) else float(sl)
+            zones.append(("stop_loss", "止损", 0, sl_price))
+        except:
+            pass
+    return zones
+
+
+def _cleanup_lock():
+    """清理进程锁文件"""
+    try:
+        os.remove("/tmp/price_monitor.lock")
+    except Exception:
+        pass
+
+def _handle_sigterm(signum, frame):
+    """收到SIGTERM时清理锁文件后退出"""
+    _cleanup_lock()
+    sys.exit(0)
+
+def run_once(round_label=""):
+    """执行一轮完整的监控流程"""
+    import os, signal  # 必须在开头import，否则os变量会被后面的局部import绑定覆盖
+    signal.signal(signal.SIGTERM, _handle_sigterm)
+    os.nice(10)  # 降低优先级，避免与DB其他写操作抢占
+    # ── 进程锁：同一时间只跑一个实例 ──
+    _lk = "/tmp/price_monitor.lock"
+    _pid = None
+    try:
+        with open(_lk) as _f:
+            _pid = int(_f.read().strip())
+        os.kill(_pid, 0)
+        print(f"[LOCK] 已有实例(PID {_pid})在运行，跳过本轮", file=sys.stderr, flush=True)
+        return
+    except (FileNotFoundError, ProcessLookupError, ValueError):
+        pass
+    with open(_lk, "w") as _f:
+        _f.write(str(os.getpid()))
+    
+    label = f" [{round_label}]" if round_label else ""
+    start = time.time()
+    TIME_BUDGET = 90  # 预留30s给输出和清理，90s内必须完成核心逻辑
+
+    # === 第一步：一次性刷新所有价格 ===
+    refreshed = refresh_data_prices()
+
+    # === 第二步：检查触发条件 ===
+    try:
+        dec = read_decisions()
+    except:
+        print(f"❌{label} 无法读取decisions(DB)", file=sys.stderr)
+        return
+
+    active = [d for d in dec.get("decisions", []) if d.get("status") == "active"]
+    state = load_state()
+    outputs = []
+    state_updated = False
+    # 时间冷却：同股同区间30分钟内不重复推
+    _push_cooldown = {}
+    _cooldown_file = "/home/hmo/.hermes/.price_push_cooldown.json"
+    try:
+        import os
+        if os.path.exists(_cooldown_file):
+            with open(_cooldown_file) as _f:
+                _push_cooldown = json.load(_f)
+    except Exception:
+        _push_cooldown = {}
+
+    def _can_push(code, zone_key):
+        now = time.time()
+        key = f"{code}_{zone_key}"
+        last = _push_cooldown.get(key, 0)
+        if now - last < 1800:  # 30分钟
+            return False
+        _push_cooldown[key] = now
+        # 持久化写入
+        try:
+            with open(_cooldown_file, "w") as _f:
+                json.dump(_push_cooldown, _f)
+        except Exception:
+            pass
+        return True
+
+    # 收集所有需要检查的代码
+    check_codes = set()
+    for d in active:
+        trig = d.get("trigger", {})
+        if trig:
+            check_codes.add(d["code"])
+
+    # 批量拉取这些股票的价格
+    prices = fetch_all_prices(list(check_codes))
+
+    for d in active:
+        code = d["code"]
+        trig = d.get("trigger", {})
+        if not trig:
+            continue
+
+        zones = get_trigger_zones(trig)
+        if not zones:
+            continue
+
+        price_info = prices.get(code)
+        if not price_info:
+            continue
+        price, _, _ = price_info
+        if price == 0:
+            continue
+
+        name = d.get("name", code)
+        if code not in state:
+            state[code] = {}
+
+        # 时间预算检查：如果超时，跳过重评只做状态记录
+        _budget_low = (time.time() - start) > TIME_BUDGET
+
+        for key, label, lo, hi in zones:
+            in_zone = lo <= price <= hi
+            prev_in_zone = state[code].get(key, None)
+
+            if in_zone and prev_in_zone != True:
+                if key == "stop_loss":
+                    outputs.append(f"⚠️ {name}({code}) {price} → 跌破止损{hi}！")
+                    record_event(code, name, "stop_loss", price, str(hi))
+                    # 止损触发 → 立即重评并推送给Dad（时间不够则直接推原始告警）
+                    if _budget_low:
+                        outputs.append(f"  📨 止损触发(超时跳过重评)→已推送Dad")
+                        if _can_push(code, "stop_loss"):
+                            push_to_xmpp(f"⚠️ {name}({code}) {price} → 跌破止损{hi}！")
+                    else:
+                        try:
+                            cost = d.get("cost", 0) or 0
+                            shares = d.get("shares", 0) or 0
+                            current_action = d.get("action", "")
+                            result = reassess_with_context(code, name, price, cost, shares, current_action)
+                            if result:
+                                timing_signal = result.get("timing_signal", "")
+                                action = result.get("action", "")
+                                if "买入" in timing_signal or "加仓" in timing_signal or timing_signal in ("卖出","止盈"):
+                                    buy_lo = d.get("entry_low", 0)
+                                    buy_hi = d.get("entry_high", 0)
+                                    rr = result.get("rr_ratio", 0)
+                                    if _can_push(code, "stop_loss"):
+                                        msg = f"🔔 {name}({code}) 价{price}→触发操作区间{max(buy_lo,0):.2f}~{buy_hi:.2f}，已触发重评|RR={rr}"
+                                        push_to_xmpp(msg)
+                                    outputs.append(f"  📨 止损重评→已推送Dad: {action}")
+                        except Exception as e:
+                            outputs.append(f"  ⚠️ 止损重评失败: {e}")
+                else:
+                    extra = ""
+                    if "_price" in key:
+                        batch_shares = trig.get(key.replace("_price", "_shares"), "")
+                        action = trig.get(key.replace("_price", "_action"), "")
+                        if batch_shares:
+                            extra = f" {action}{batch_shares}股" if action else f" {batch_shares}股"
+                    elif key in ("take_profit_zone",):
+                        act = trig.get("take_profit_action", "")
+                        if act:
+                            extra = f"（{act}）"
+                    outputs.append(f"⚡ {name}({code}) {price} → 进入{label}{lo}~{hi}{extra}")
+                    record_event(code, name, "entry_zone", price, f"{lo}~{hi}", label)
+                    # 进入区间 → 立即重评并推送给Dad（时间不够则跳过重评直接推原始告警）
+                    if _budget_low:
+                        if _can_push(code, key):
+                            push_to_xmpp(f"⚡ {name}({code}) {price} → 进入{label}{lo}~{hi}")
+                        outputs.append(f"  📨 区间触发(超时跳过重评)→已推送Dad")
+                    else:
+                        try:
+                            cost = d.get("cost", 0) or 0
+                            shares = d.get("shares", 0) or 0
+                            current_action = d.get("action", "")
+                            result = reassess_with_context(code, name, price, cost, shares, current_action)
+                            if result:
+                                timing_signal = result.get("timing_signal", "")
+                                action = result.get("action", "")
+                                # 格式化区间描述（止盈区lo=0时美化显示）
+                                if key == "take_profit_zone" and lo == 0:
+                                    zone_desc = f"止盈监控(目标{hi:.0f})"
+                                else:
+                                    zone_desc = f"操作区间{lo}~{hi}"
+                                if "买入" in timing_signal or "加仓" in timing_signal or timing_signal in ("卖出","止盈"):
+                                    rr = result.get("rr_ratio", 0)
+                                    if _can_push(code, key):
+                                        msg = f"🔔 {name}({code}) 价{price}→触发{zone_desc}，已触发重评|RR={rr}"
+                                        push_to_xmpp(msg)
+                                        outputs.append(f"  📨 区间触发重评→已推送Dad: {action}")
+                                else:
+                                    reason = f"重评结果:{timing_signal}，不构成操作建议"
+                                    outputs.append(f"  📋 本地日志(不推): {reason}")
+                        except Exception as e:
+                            outputs.append(f"  ⚠️ 区间重评失败: {e}")
+                state[code][key] = True
+                state_updated = True
+
+            elif not in_zone and prev_in_zone == True:
+                if key != "stop_loss":
+                    outputs.append(f"📌 {name}({code}) {price} → 离开{label}{lo}~{hi}")
+                state[code][key] = False
+                state_updated = True
+
+    # === 第三步：买入区偏离检测 + 自动重评 ===
+    reassesed_codes = []
+    # 先做急跌检测（仅持仓，自选股不推送暴跌告警）
+    holdings_codes = {d["code"] for d in active if d.get("shares", 0) > 0}
+    for d in active:
+        code = d["code"]
+        # 非持仓跳过
+        if code not in holdings_codes:
+            continue
+        name = d.get("name", code)
+        price_info = prices.get(code)
+        if not price_info:
+            continue
+        price, _, change_pct = price_info
+        if price == 0:
+            continue
+        # 单日跌幅>7%告警（不依赖zone边界，盘中急跌即触发）
+        try:
+            cp = float(change_pct) if change_pct else 0
+        except:
+            cp = 0
+        if cp <= -7:
+            prev_alert = state.get(code, {}).get("__sharp_decline_triggered", False)
+            if not prev_alert:
+                stop_loss = d.get("stop_loss", 0)
+                sl_note = f" 止损{stop_loss}" if stop_loss else ""
+                msg = f"🔻 {name}({code}) {price} 暴跌{cp:.1f}%！{sl_note}"
+                push_to_xmpp(msg)
+                outputs.append(msg)
+                state.setdefault(code, {})["__sharp_decline_triggered"] = True
+                state_updated = True
+                # 立即持久化，防止后续超时导致状态丢失而重复推送
+                save_state(state)
+        elif cp > -5:
+            # 反弹后清除告警标记，下次再跌还能报
+            state.setdefault(code, {}).pop("__sharp_decline_triggered", None)
+
+    for d in active:
+        code = d["code"]
+        name = d.get("name", code)
+        price_info = prices.get(code)
+        if not price_info:
+            continue
+        price, _, _ = price_info
+        if price == 0:
+            continue
+
+        # 从 decisions (DB holding_strategies) 中读取 analysis 的买入区
+        entry_low = d.get("entry_low", 0)
+        entry_high = d.get("entry_high", 0)
+        if not entry_low or not entry_high:
+            continue
+
+        in_buy_zone = entry_low <= price <= entry_high
+        prev_in_buy_zone = state.get(code, {}).get("__buy_zone", None)
+
+        # 状态变化时才触发
+        if in_buy_zone and prev_in_buy_zone == False:
+            # 重新进入买入区 → 重评确认区间是否仍然有效
+            outputs.append(f"🔄 {name}({code}) {price} → 重新进入买入区{entry_low}~{entry_high}，触发技术面重评")
+            do_reassess = True
+        elif not in_buy_zone and prev_in_buy_zone == True:
+            # 离开买入区 → 立即重评，更新止损/止盈/区间
+            outputs.append(f"🔄 {name}({code}) {price} → 离开买入区{entry_low}~{entry_high}，立即技术面重评")
+            do_reassess = True
+        else:
+            do_reassess = False
+
+        if do_reassess and HAS_REASSESS:
+            try:
+                cost = d.get("cost", 0) or 0
+                shares = d.get("shares", 0) or 0
+                profit_pct = (price - cost) / cost * 100 if cost else 0
+                is_deep_loss = profit_pct < -20
+                sentiment = "neutral"
+                if d.get("tech_snapshot"):
+                    if "bearish" in d["tech_snapshot"]:
+                        sentiment = "bearish"
+                    elif "bullish" in d["tech_snapshot"]:
+                        sentiment = "bullish"
+
+                # 调用技术面驱动重评（非机械百分比）
+                result = reassess_strategy(
+                    code, name, price, cost, shares,
+                    current_action=d.get("action", ""),
+                    volume_signal="中性", sentiment=sentiment,
+                )
+                outputs.append(f"  📊 新策略: 损{result['stop_loss']} 盈{result['take_profit']} 区{result['entry_low']}~{result['entry_high']} RR={result['rr_ratio']}")
+                reassesed_codes.append(code)
+            except Exception as e:
+                outputs.append(f"  ⚠️ 重评失败: {e}")
+
+        # 更新买入区状态
+        if "__buy_zone" not in state.get(code, {}):
+            if code not in state:
+                state[code] = {}
+        state[code]["__buy_zone"] = in_buy_zone
+        state_updated = True
+
+    # 如果有重评过的股票，更新 DB holding_strategies（此前写入 decisions.json，已废弃）
+    if reassesed_codes and HAS_REASSESS:
+        # ── 5分钟冷却：regenerate_all 开销太大，不每2分钟跑一次 ──
+        _regen_marker = "/tmp/price_monitor_regen_at"
+        _skip_regen = False
+        try:
+            if os.path.exists(_regen_marker):
+                with open(_regen_marker) as _f:
+                    _last_regen = float(_f.read().strip())
+                if time.time() - _last_regen < 300:
+                    _skip_regen = True
+        except:
+            pass
+
+        if _skip_regen:
+            outputs.append(f"  ⏭ 跳过全量重评（距上次<5min），下次再跑")
+        else:
+            try:
+                from strategy_lifecycle import regenerate_all
+                r = regenerate_all(stdout=False)
+                outputs.append(f"  ✅ 策略已全量重评: {r.get('ok',0)}/{r.get('total',0)}成功")
+                outputs.append(f"  📌 触发股票: {', '.join(reassesed_codes)}")
+                try:
+                    with open(_regen_marker, "w") as _f:
+                        _f.write(str(time.time()))
+                except:
+                    pass
+            except Exception as e:
+                outputs.append(f"  ⚠️ 全量重评失败: {e}")
+
+    # === 第四步：输出 ===
+    now_str = datetime.now().strftime("%H:%M:%S")
+    elapsed = time.time() - start
+
+    if outputs:
+        print(f"\n🔔 {now_str}{label}")
+        for o in outputs:
+            print(o)
+        print(f"\n<structured_data>{json.dumps({'type':'价格监控','time':now_str,'triggers':outputs}, ensure_ascii=False)}</structured_data>")
+    else:
+        # 无触发时 SILENT（中继不推送）
+        print(f"[SILENT]{label} 价格正常 | {refreshed}只已刷新 | {elapsed:.1f}s")
+
+    if state_updated:
+        save_state(state)
+
+    # 输出耗时
+    print(f"⏱{label} {elapsed:.1f}s", flush=True)
+
+    # 清理进程锁
+    try:
+        os.remove("/tmp/price_monitor.lock")
+    except Exception:
+        pass
+
+
+def main():
+    """每cron触发跑一轮"""
+    run_once()
+
+
+if __name__ == "__main__":
+    main()