#!/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
from alert_logger import record_alert, clear_alert
from messenger import send as messenger_send

# ── 温区感知 + 信号溯源（2026-08-13）──
try:
    from regime_gate import _load_weights
except Exception:
    _load_weights = None

def _load_active_strats():
    """当前温区激活策略集合（strategy_weights.json matched）"""
    if _load_weights is None:
        return None
    try:
        w = _load_weights()
        if w and w.get("weights"):
            return {k for k, v in w["weights"].items() if v.get("matched")}
    except Exception:
        pass
    return None

def _is_strategy_active(d_entry):
    """该持仓策略是否在当前温区激活。未标记策略默认激活（兼容持仓管理/风控）"""
    act = _load_active_strats()
    if act is None:
        return True
    for key in ("tag", "version", "strategy_type", "decision_type"):
        v = d_entry.get(key)
        if v and str(v) in act:
            return True
    # 该策略在权重表中但不在激活集 → 非激活
    try:
        w = _load_weights() or {}
        wmap = w.get("weights") or {}
        for key in ("tag", "version", "strategy_type", "decision_type"):
            v = d_entry.get(key)
            if v and v in wmap:
                return v in act
    except Exception:
        pass
    return True

def _record_signal(code, name, reason, strategy=""):
    """信号溯源：记录到 signal_ledger（温区/温度按标的市场，2026-08-14 阶段4）"""
    try:
        from signal_ledger import record_signal
        from market_config import get_regime_temp
        _regime, _temp = get_regime_temp(code)
        record_signal(code=code, name=name, strategy=strategy, version=strategy,
                      regime=_regime, temp_band=_temp,
                      reason=reason, source_module="price_monitor")
    except Exception:
        pass

BREACH_PATH = "/home/hmo/.hermes/zone_breach.json"
STATE_PATH = "/home/hmo/.hermes/price_trigger_state.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

# 市场抽象层（阶段1：行情符号/市场判断唯一事实源）
from market_config import kline_symbol, market_for_code

# ── 消息通道统一路由(broadcast/xmpp by delivery) ──
try:
    from messenger import install_stdio_hook as _msh
    _msh()
except Exception:
    pass

# 策略重评依赖（技术面驱动，非机械百分比）
sys.path.insert(0, "/home/hmo/web-dashboard")
try:
    from strategy_lifecycle import reassess_strategy, reassess_with_context
    import subprocess as _sp2
    _REASSESS_SCRIPT = "/home/hmo/MoFin/deploy/profile-scripts/per_stock_reassess.py"
    _REASSESS_OLD = "/home/hmo/MoFin/data/.price_reassess_last.json"

    def _done_reassess_recently(code):
        """冷却检查：同票 30 分钟内已触发过 12维重评 → 跳过（防每2分钟重复烧LLM）"""
        try:
            import os as _os
            if _os.path.exists(_REASSESS_OLD):
                _last = json.load(open(_REASSESS_OLD, encoding="utf-8"))
                if code in _last and time.time() - _last[code] < 1800:
                    return True
        except Exception:
            pass
        return False

    def _mark_reassess(code):
        """记录重评时间（冷却标记）"""
        try:
            import os as _os
            _last = {}
            if _os.path.exists(_REASSESS_OLD):
                try:
                    _last = json.load(open(_REASSESS_OLD, encoding="utf-8"))
                except Exception:
                    _last = {}
            _last[code] = time.time()
            json.dump(_last, open(_REASSESS_OLD, "w"), ensure_ascii=False)
        except Exception:
            pass

    def _do_llm_reassess(code, name, price, cost, shares, current_action):
        """真正12维LLM重评（per_stock_reassess.py）。返回最新参数的dict或None。"""
        if _done_reassess_recently(code):
            print(f"  ⏭ {code} 30分钟内已重评，跳过（冷却）", flush=True)
            return None
        try:
            _r = _sp2.run(
                [sys.executable, _REASSESS_SCRIPT, code],
                capture_output=True, text=True, timeout=600  # 2026-08-24 180→600: 子进程含collect_data+LLM(300s上限),并发排队下给足
            )
            _mark_reassess(code)
            # 2026-08-24 防鬼消息：子进程失败不得继续读旧策略冒充"重评结论"
            if _r.returncode != 0:
                print(f"  ⚠️ {code} 12维重评子进程失败(rc={_r.returncode}): {(_r.stderr or _r.stdout or '')[-300:]}", file=sys.stderr, flush=True)
                return None
        except Exception as e:
            print(f"  ⚠️ {code} 12维重评异常: {e}", file=sys.stderr)
            return None
        # 重评后从DB读最新参数（子进程已写库）
        try:
            import sqlite3 as _sq
            _c = _sq.connect("/home/hmo/MoFin/data/mofin.db", timeout=30)
            _c.row_factory = _sq.Row
            _row = _c.execute(
                "SELECT timing_signal, action, stop_loss, take_profit, entry_low, entry_high, rr_ratio, reassessed_at "
                "FROM holding_strategies WHERE code=? AND status='active'", (code,)).fetchone()
            _c.close()
            # 2026-08-24 防鬼消息：rc=0但LLM失败时策略未刷新（reassessed_at不变），
            # 旧策略不得冒充"重评结论"——reassessed_at 超过10分钟视为重评未生效
            if _row:
                from datetime import datetime as _dt
                _ra = _row["reassessed_at"]
                if not _ra or (_dt.now() - _dt.fromisoformat(str(_ra))).total_seconds() > 600:
                    print(f"  ⚠️ {code} 重评后策略未刷新(reassessed_at={_ra})，按重评失败处理", file=sys.stderr, flush=True)
                    return None
            if _row:
                return {
                    "timing_signal": _row["timing_signal"] or "",
                    "action": _row["action"] or "",
                    "stop_loss": _row["stop_loss"] or 0,
                    "take_profit": _row["take_profit"] or 0,
                    "entry_low": _row["entry_low"] or 0,
                    "entry_high": _row["entry_high"] or 0,
                    "rr_ratio": _row["rr_ratio"] or 0,
                }
        except Exception as e:
            print(f"  ⚠️ {code} 读重评结果失败: {e}", file=sys.stderr)
        return None
    HAS_REASSESS = True
except ImportError:
    HAS_REASSESS = False


# ── 2026-08-24 并发重评：单只"重评+结果分发+推送"（ThreadPoolExecutor 6 路调用）──
_can_push_lock = __import__("threading").Lock()

def _reassess_one(ctx):
    """并发执行单只 12 维重评并按结论推送。返回值: dict(outputs/zone_entries/reassessed) 或 None。
    取自原 zones 循环内两个调用点(止损/区间分支)的重评+推送逻辑,代价是共享 _can_push 需加锁。"""
    code = ctx["code"]; name = ctx["name"]; price = ctx["price"]; d = ctx["d"]
    key = ctx["key"]; hi = ctx["hi"]; lo = ctx["lo"]
    out, ze = [], []
    try:
        cost = d.get("cost", 0) or 0
        shares = d.get("shares", 0) or 0
        current_action = d.get("action", "")
        result = _do_llm_reassess(code, name, price, cost, shares, current_action)
        rea = []
        if key != "stop_loss" and result:
            rea.append(code)
        elif key == "stop_loss":
            rea.append(code)
        if result:
            timing_signal = result.get("timing_signal", "")
            action = result.get("action", "")
            rr = result.get("rr_ratio", 0)
            sl_new = result.get("stop_loss", 0)
            tp_new = result.get("take_profit", 0)
            if key == "stop_loss":
                # ── 止损分支推送 ──
                reason_extra = ""
                for k in ("action_note", "signal_factors"):
                    v = result.get(k)
                    if v:
                        if isinstance(v, list):
                            v = "、".join(str(x) for x in v)
                        reason_extra += f" [{k}={v}]"
                if timing_signal in ("卖出", "止盈") and (d.get("shares") or 0) > 0:
                    with _can_push_lock:
                        _cp = _can_push(code, "stop_loss")
                    if _cp:
                        msg = (f"🔔 {name}({code}) 价{price} → 跌破止损，重评结论【{timing_signal}】| RR={rr}"
                               f" | 止损{sl_new}/止盈{tp_new}"
                               f" | 操作: {str(action)[:250]}{reason_extra}")
                        _push_action("操作信号", msg)
                elif "持有" in timing_signal or "关注" in timing_signal or "观望" in timing_signal:
                    try:
                        import sqlite3 as _sq
                        _c = _sq.connect("/home/hmo/MoFin/data/mofin.db", timeout=10)
                        if sl_new:
                            _c.execute(
                                "UPDATE holding_strategies SET stop_loss=?, updated_at=datetime('now','localtime') "
                                "WHERE code=? AND status='active'", (sl_new, code))
                        _c.commit()
                        _c.close()
                    except Exception:
                        pass
                    ze.append(f"{name}({code}) {price}→破止损但重评{timing_signal}，止损已更新至{sl_new}")
                    out.append(f"  📨 破止损→重评{timing_signal}（不卖出，止损更新{sl_new}）: {str(action)[:120]}")
                else:
                    ze.append(f"{name}({code}) {price}→入区+重评{timing_signal}|RR={rr}")
                    out.append(f"  📨 止损重评→{timing_signal}: {str(action)[:120]}")
            else:
                # ── 区间分支推送 ──
                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 ("卖出", "止盈"):
                    if timing_signal in ("卖出", "止盈") and (d.get("shares") or 0) > 0:
                        with _can_push_lock:
                            _cp = _can_push(code, key)
                        if _cp:
                            msg = f"🔔 {name}({code}) 价{price}→触发{zone_desc}，已触发重评|RR={rr}"
                            _push_action("操作信号", msg)
                        out.append(f"  📨 区间触发重评→已推送Dad: {action}")
                        try:
                            _record_signal(code, name, f"风控:{timing_signal} {zone_desc}", strategy=d.get("tag") or d.get("version") or "")
                        except Exception:
                            pass
                    else:
                        if _is_strategy_active(d):
                            ze.append(f"{name}({code}) {price}→{zone_desc}+重评{timing_signal}|RR={rr}")
                            out.append(f"  📋 机会记入摘要: {timing_signal} RR={rr}")
                            try:
                                _record_signal(code, name, f"买入机会:{zone_desc} {timing_signal}", strategy=d.get("tag") or d.get("version") or "")
                            except Exception:
                                pass
                        else:
                            out.append(f"  🧊 温区未激活跳过: {name}({code}) {timing_signal} RR={rr}")
                else:
                    reason = f"重评结果:{timing_signal}，不构成操作建议"
                    out.append(f"  📋 本地日志(不推): {reason}")
        else:
            # 重评未生效（冷却/子进程失败）→ 原始告警+人工核查标注(不拿旧策略冒充结论)
            with _can_push_lock:
                _cp = _can_push(code, "stop_loss" if key == "stop_loss" else key)
            if _cp:
                if key == "stop_loss":
                    _push_action("止损告警", f"⚠️ {name}({code}) {price} → 跌破止损{hi}！（12维重评未生效，暂无最新结论，请人工核查）")
                else:
                    _push_action("止损告警", f"⚠️ {name}({code}) {price} → 触发区间{lo}~{hi}但重评未生效，请人工核查")
            out.append(f"  ⚠️ {code} 重评未生效，已发原始告警")
        return {"outputs": out, "zone_entries": ze, "reassessed": rea}
    except Exception as e:
        out.append(f"  ⚠️ {code} 重评失败: {e}")
        try:
            record_alert(level="error", source="price_monitor", title="重评失败", detail=str(e)[:200], code=code)
        except Exception:
            pass
        return {"outputs": out, "zone_entries": ze, "reassessed": []}

UA = "Mozilla/5.0"

# ── XMPP推送 ──────────────────────────────────────────────────────────
XMPP_USER = "hmo@yoin.fun"
XMPP_BRIDGE = "http://127.0.0.1:5805/"

def push_to_xmpp(text):
    """原始直推（已废弃直用）——保留给极少数必须原样的场景。
    新代码请用 _push_action/_push_digest。"""
    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)


# ── 分级推送（2026-07-21 信噪比纪律，红线#12）──
# ACTION: 破止损/重评确认的操作信号 — 直通不限速
# INFO:   未确认的进区提示 — 聚合成摘要，30min 限 1 条
def _push_action(category, text):
    # 2026-08-24 老莫新规则：只有含"操作建议"的消息才上 XMPP（操作信号/推荐）；
    # 无操作建议的异动播报/止损告警/急跌告警只进 broadcast，不打扰
    _XMPP_CATS = ("操作信号", "重点推荐", "推荐操作")
    _is_advice = category in _XMPP_CATS
    try:
        messenger_send(title=f"MoFin持仓异动", content=text, source="price_monitor.py",
                       channel="xmpp" if _is_advice else "broadcast")
    except Exception as e:
        print(f"[messenger失败] {e}", file=sys.stderr)
    if _is_advice:
        try:
            from alert_helper import notify, ACTION
            notify(category, text, ACTION)
        except Exception as e:
            print(f"[ACTION推送失败] {e}", file=sys.stderr)


def _push_digest(category, text):
    # 普通监控信息：broadcast 归档（不打扰 xmpp）
    try:
        messenger_send(title=f"MoFin价格监控", content=text, source="price_monitor.py")
    except Exception as e:
        print(f"[messenger归档失败] {e}", file=sys.stderr)
    try:
        from alert_helper import notify, INFO
        notify(category, text, INFO)
    except Exception as e:
        print(f"[INFO推送失败] {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()
        sym = kline_symbol(code_s)
        if sym is None:
            sym = f"hk{code_s}"  # 兜底：非5/6位数字 → 沿用旧 else 分支 hk 前缀
        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)
        try:
            record_alert(level="warning", source="price_monitor", title="批量拉取失败", detail=str(e)[:200])
        except Exception:
            pass
        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)
        try:
            record_alert(level="warning", source="price_monitor", title="读取持仓失败", detail=str(e)[:200])
        except Exception:
            pass
        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)
                        )

                # 实时技术指标联动更新（2026-08-24 整体预算10s：每只默认5s超时×114只=必撞alarm，
                # 撞锁时快速跳过，指标晚一轮更新无伤，zones止损判定才是命脉）
                try:
                    from realtime_indicators import update_single
                    _ri_start = time.time()
                    for h in db_holdings:
                        if time.time() - _ri_start > 10:
                            break
                        _ri_code = h.get("code", "")
                        _ri_price = h.get("price", 0)
                        if _ri_code and _ri_price:
                            update_single(_ri_code, float(_ri_price))
                except Exception as e:
                    print("  realtime_indicators error:", e, file=sys.stderr)
                # 补充策略股/自选股的价格（不在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)
                try:
                    record_alert(level="error", source="price_monitor", title="DB同步异常", detail=str(e)[:200])
                except Exception:
                    pass
                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 record_event(code, name, event_type, price, trigger_value, event_label=""):
    """记录一次价格触发事件到 DB price_events 表（唯一权威存储，JSON 已退役）。

    price_events.code 有 FK -> stocks(code)，未注册的股票（新候选/港股）
    先注册再写事件，否则 FK 失败事件丢失。
    """
    now = datetime.now().isoformat()

    if HAS_DB:
        try:
            from mofin_db import get_conn, write_price_event
            _c = get_conn()
            # 市场判断统一走 market_for_code（港股=5位0/1开头）；A股再按 6/9 区分沪深
            if market_for_code(code) == 'hk':
                _exch, _typ = ("HK", "H")
            elif str(code).startswith(("6", "9")):
                _exch, _typ = ("SH", "A")
            else:
                _exch, _typ = ("SZ", "A")
            # 2026-08-22 名称治理：无名字先查 stocks 已有真名，再腾讯 quote 兜底，杜绝 name=code 垃圾行
            if not name:
                try:
                    _r = _c.execute("SELECT name FROM stocks WHERE code=?", (str(code),)).fetchone()
                    if _r and _r[0] and _r[0] != str(code):
                        name = _r[0]
                except Exception:
                    pass
            if not name:
                try:
                    from mofin_db import fetch_stock_name_tencent
                    name = fetch_stock_name_tencent(code) or ""
                except Exception:
                    pass
            _c.execute("INSERT OR IGNORE INTO stocks (code, name, exchange, type, updated_at) VALUES (?,?,?,?,?)",
                       (str(code), name or str(code), _exch, _typ, now))
            if name:
                # 治愈已有 name=code 的坏行（INSERT OR IGNORE 不会更新已存在行）
                _c.execute("UPDATE stocks SET name=?, updated_at=? WHERE code=? AND name=?",
                           (name, now, str(code), str(code)))
            _c.commit()
            write_price_event(_c, code=code, name=name, event_type=event_type,
                              price=round(price, 2), trigger_value=str(trigger_value),
                              event_label=event_label)
            _c.close()
        except Exception as e:
            print(f"[price_events DB写入失败] {e}", file=sys.stderr)


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 _handle_sigalrm(signum, frame):
    """收到SIGALRM强制超时时清理锁文件后退出"""
    _cleanup_lock()
    print(f"[TIMEOUT] 本轮执行超时({signum}s)，已清理锁文件", file=sys.stderr, flush=True)
    sys.exit(0)

def run_once(round_label=""):
    """执行一轮完整的监控流程"""
    import os, signal  # 必须在开头import，否则os变量会被后面的局部import绑定覆盖
    signal.signal(signal.SIGTERM, _handle_sigterm)
    signal.signal(signal.SIGALRM, _handle_sigalrm)
    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()))
    signal.alarm(600)  # 2026-08-24 120→300→600：重评并发后一轮最多6只LLM×~90s,600s覆盖整轮
    
    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 = []
    _reassess_pending = []  # 2026-08-24 并发重评队列表(循环内只快判收集,循环后统一并发)
    reassesed_codes = []  # 2026-08-25 挪到并发段引用之前（919行extend引用在937行定义之前→UnboundLocalError,price_monitor每次触发就崩）
    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", {})
        # 2026-08-24 兜底：trig 空但主层有止损/区间/止盈的版本化策略也纳入
        # （否则 prices 里没它们,zones 循环 price_info=None 直接 continue——603920 盲区第二环）
        if trig or d.get("stop_loss") or (d.get("entry_low") and d.get("entry_high")) or d.get("take_profit"):
            check_codes.add(d["code"])

    # 批量拉取这些股票的价格
    prices = fetch_all_prices(list(check_codes))

    # 本轮进区事件收集（聚合成一条摘要推送，替代逐条轰炸）
    _zone_entries = []

    for d in active:
        code = d["code"]
        trig = d.get("trigger", {}) or {}

        zones = get_trigger_zones(trig)
        # ── 2026-08-24 监控盲区根治：trigger_json 为空的版本化策略（76/112!），
        # 用主层字段构建基础 zones（stop_loss/entry_zone/take_profit_zone）。
        # 否则这些股票完全脱离止损/止盈/买入区监控（603920 破止损无告警事件的根因）。
        try:
            if d.get("stop_loss") and not any(z[0] == "stop_loss" for z in zones):
                zones.append(("stop_loss", "止损", 0, float(d["stop_loss"])))
            if d.get("entry_low") and d.get("entry_high") and not any(z[0] == "entry_zone" for z in zones):
                zones.append(("entry_zone", "买入区", float(d["entry_low"]), float(d["entry_high"])))
            if d.get("take_profit") and not any(z[0] == "take_profit_zone" for z in zones):
                _tp = float(d["take_profit"])
                zones.append(("take_profit_zone", "止盈区间", _tp * 0.995, _tp * 100))
        except Exception:
            pass
        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] = {}

        # ── v_combo波段出场（2026-08-13 已移除硬编码）──
        # 老莫原则：k线形态判断在LLM提示词上体现（batch_reassess 提示词已含波段出场形态），
        # 不在代码里硬编码"连续2日收破MA10"特定规则。
        # swing持仓的 stop_loss/entry_zone/take_profit_zone 已由通用 zones 循环触发 LLM 重评，
        # timing_signal 含"卖出/止盈"才发推荐 → 统一"进区→重评→维持才发"。

        # 时间预算检查：如果超时，跳过重评只做状态记录
        _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_action("止损告警", f"⚠️ {name}({code}) {price} → 跌破止损{hi}！")
                    else:
                        # 2026-08-24 并发化：不串行等重评,收集到pending,循环后6路并发(OCG router 6 key)
                        _reassess_pending.append({
                            "code": code, "name": name, "price": price, "d": d,
                            "key": key, "label": label, "lo": lo, "hi": hi, "trig": trig})
                        outputs.append(f"  📨 {code} 破止损→重评入队(并发)")
                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):
                            _zone_entries.append(f"{name}({code}) {price}→{label}{lo}~{hi}")
                        outputs.append(f"  📨 区间触发(超时)→记入摘要")
                    else:
                        # 2026-08-24 并发化(同止损分支)
                        _reassess_pending.append({
                            "code": code, "name": name, "price": price, "d": d,
                            "key": key, "label": label, "lo": lo, "hi": hi, "trig": trig})
                        outputs.append(f"  📨 {code} 区间触发→重评入队(并发)")
                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

    # === 第二步收尾：进区事件只记日志不推送（2026-07-22 老爸：进区提示=噪音，关停）===
    # 正式推荐由 12维→tag→摘要队列 通道负责；破止损/卖出/止盈(持仓)仍走 ACTION 直推。
    if _zone_entries:
        outputs.append(f"📋 进区事件{len(_zone_entries)}只（仅记日志不推送）: {'; '.join(_zone_entries[:5])}")

    # ── 2026-08-24 并发重评（老莫：OCG router 6 key，为什么不并发！）──
    # 触发已全部快判完成,这里6路并发跑LLM重评+结果推送
    if _reassess_pending:
        import concurrent.futures as _cf2
        outputs.append(f"  ⏭ {len(_reassess_pending)} 只重评入队,6路并发(ocg_router 6key)")
        _ex = _cf2.ThreadPoolExecutor(max_workers=6)
        try:
            for _sub in _ex.map(_reassess_one, _reassess_pending):
                if _sub:
                    outputs.extend(_sub.get("outputs", []))
                    _zone_entries.extend(_sub.get("zone_entries", []))
                    reassesed_codes.extend(_sub.get("reassessed", []))
        finally:
            _ex.shutdown(wait=True)

    # === 第三步：买入区偏离检测 + 自动重评 ===
    # 2026-08-11 修复：告警/重评只在连续竞价时段执行（避开集合竞价虚拟撮合价误报）。
    # 2026-08-14 港股接入：时段判断按市场——A股 9:30-11:30/13:00-15:00，港股 9:30-12:00/13:00-16:00。
    # 港股16:00收盘，15:00-16:00 仍需监控持仓（A股已收盘价格不变，不会误报）。
    from market_config import is_trading_now
    _now_hm = datetime.now().strftime("%H:%M")
    _trading = is_trading_now('a') or is_trading_now('hk')
    if not _trading:
        print(f"[时段] {_now_hm} 非连续竞价时段（集合竞价/午休/盘后），跳过告警与重评（价格已刷新）", flush=True)
        # 仍提交状态（防止告警标记滞留）
        if state_updated:
            save_state(state)
        return


    # 先做急跌检测（仅持仓，自选股不推送暴跌告警）
    holdings_codes = set()
    for d in active:
        shares = d.get("shares", 0)
        if isinstance(shares, (int, float)):
            if shares > 0:
                holdings_codes.add(d["code"])
        else:
            # 非数值shares（如被错误写入的字符串），兜底处理
            holdings_codes.add(d["code"])
            print(f"  [WARN] {d.get('code')} shares为非数值({shares!r})，视为持仓处理", flush=True)
    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_action("急跌告警", 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 = _do_llm_reassess(code, name, price, cost, shares, d.get("action", ""))
                if result:
                    outputs.append(f"  📊 新策略: 损{result['stop_loss']} 盈{result['take_profit']} 区{result['entry_low']}~{result['entry_high']} RR={result['rr_ratio']}")
                    reassesed_codes.append(code)
                else:
                    outputs.append(f"  ⏭ {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)

    # 取消超时定时器（正常完成）
    signal.alarm(0)

    # ── 策略追踪评估（2026-07-27 老爸：检查推荐是否触发止盈/止损）──
    try:
        import mofin_db
        _conn = mofin_db.get_conn()
        tracked = mofin_db.check_strategy_outcomes(_conn)
        _conn.close()
        if tracked:
            print(f"  [TRACK] {tracked}条推荐触发止盈/止损", flush=True)
    except Exception as e:
        print(f"  [TRACK] 检查失败: {e}", flush=True)

    # 清理进程锁
    try:
        os.remove("/tmp/price_monitor.lock")
    except Exception:
        pass


def _singleton_guard(max_age_sec=300):
    """自愈式单例守卫（2026-08-05 进程堆积事故修复）
    背景：cron每2分钟拉起本脚本，外网抖动时实例挂起不退，timeout(仅SIGTERM)杀不动，
    5个并发烧~155% CPU。规则：
    - 有其他"新鲜"实例（存活<max_age）在跑 → 本实例立即退出（防堆积）
    - 有"卡死"实例（存活>max_age）→ SIGKILL 接管（自愈）
    """
    import subprocess as _sp, os as _os, sys as _sys
    my_pid = _os.getpid()
    try:
        out = _sp.run(["ps", "-C", "python3", "-o", "pid,etimes,cmd"],
                      capture_output=True, text=True, timeout=10).stdout
        for line in out.splitlines():
            if "price_monitor.py" not in line:
                continue
            parts = line.split(None, 2)
            if len(parts) < 3:
                continue
            try:
                pid = int(parts[0]); age = int(parts[1])
            except ValueError:
                continue
            if pid == my_pid:
                continue
            if age > max_age_sec:
                try:
                    _os.kill(pid, 9)
                    print(f"[guard] SIGKILL卡死实例 pid={pid} age={age}s", flush=True)
                except ProcessLookupError:
                    pass
            else:
                print(f"[guard] 已有新鲜实例 pid={pid} age={age}s 在跑, 本实例退出", flush=True)
                _sys.exit(0)
    except Exception as _e:
        print(f"[guard] 守卫异常(放行): {_e}", flush=True)


def main():
    """每cron触发跑一轮"""
    _singleton_guard(max_age_sec=300)
    run_once()


if __name__ == "__main__":
    main()