fix(price_monitor): DB写锁死锁根治 - 统一BEGIN IMMEDIATE + 5次重试指数退避 + emergency WAL checkpoint

- 统一一个BEGIN IMMEDIATE事务包裹所有写操作，替代独立写函数调用 - 5次重试 + 指数退避 (1s,2s,4s,8s,16s)，原3次+固定2s - 重试耗尽后自动 emergency WAL checkpoint(TRUNCATE)释放死锁 - try/except确保连接始终释放，修复conn泄漏 - 三副本同步：profile/scripts + MoFin/scripts + MoFin/root
2026-07-14 10:30:15 +08:00
parent 9239ab40c0
commit 02d1c93923
4 changed files with 269 additions and 82 deletions
@@ -260,3 +260,16 @@ Bot被bash包装器手动启动（非systemd），形成孤儿进程。systemd
 - 每日汇总只发一次(16:35)
 - 健康监控新增数据实体检查
 - **涉及文件**: technical_analysis.py, mofin_health.py, cron_to_xmpp.py, cron_health_monitor.py
+
+## 2026-07-14 — price_monitor DB写锁死锁根治
+
+### 根因
+多 cron 脚本（price_monitor、mofin_health 等）同时 BEGIN IMMEDIATE 写 mofin.db，WAL 文件膨胀到 2.7MB 未 checkpoint，进程卡在 D 状态，后续全部超时。原代码的 3 次重试 + 2s 退避不足，且 write_portfolio_summary / write_live_prices 返回值未检查。
+
+### 修复
+1. **统一 BEGIN IMMEDIATE + 内联 SQL**（替代调用 write_holdings_batch/write_portfolio_summary/write_live_prices）：一个事务包裹所有写操作，任一失败立即 rollback + 重试
+2. **5 次重试 + 指数退避**（1s → 2s → 4s → 8s → 16s，共 ~31s），原 3 次 + 固定 2s
+3. **Emergency WAL checkpoint**：所有重试耗尽后自动执行 PRAGMA wal_checkpoint(TRUNCATE) 释放死锁
+4. **try/except 确保连接始终释放**（rollback + close），修复原代码异常时 conn 泄漏
+5. **三副本同步**：profile/scripts + MoFin/scripts + MoFin/root
+- **涉及文件**: scripts/price_monitor.py, /home/hmo/MoFin/scripts/price_monitor.py, /home/hmo/MoFin/price_monitor.py
@@ -149,3 +149,38 @@ bash包装器启动bot会绕开systemd管理，导致：
 - 每日汇总只发一次（16:35 + sent标记防重复）
 - cron_health_monitor新增数据实体完整性检查
 - 涉及文件: scripts/technical_analysis.py, scripts/strategy_lifecycle.py, scripts/mofin_health.py, scripts/cron_to_xmpp.py, scripts/cron_health_monitor.py
+
+## 2026-07-14 — 系统检查：Dad未收到任何消息的诊断
+
+### 发现问题
+1. **cron_to_xmpp.py Bug**: body提取`content.split("## Response")`用`parts[1]`取到的是skill内容中的 ## Response（第一个），而不是agent实际回复的 ## Response（最后一个）。导致开盘简报等LLM报告的正文取错，被误判为[SILENT]静默拦截。
+2. **price_monitor.py SQLite写锁死锁**: 多个cron脚本（price_monitor + mofin_health等）同时BEGIN IMMEDIATE写 mofin.db，进程卡在D状态，WAL文件膨胀到2.7MB，后续全部超时。
+3. **mofin_db.py版本不同步**: cron版本和MoFin版本有差异（cron版本多了保留full_analysis/reassessed_at的代码）。
+
+### 修改了什么
+- `cron_to_xmpp.py` — 改 `parts[1]` → `parts[-1]`（取最后一个Response节）
+- `mofin_db.py` (MoFin目录) — 同步cron版本的reassessed_at列处理和full_analysis保留逻辑
+- `price_monitor.py` — 添加debug时序追踪，临时kill D状态进程+WAL checkpoint
+- `.silent_daily_count.json` — 纠正静默计数（-1因为开盘简报被误拦）
+
+### 文件
+- Modified: /home/hmo/.hermes/profiles/position-analyst/scripts/cron_to_xmpp.py
+- Modified: /home/hmo/MoFin/scripts/mofin_db.py
+- Modified: /home/hmo/.hermes/profiles/position-analyst/scripts/price_monitor.py
+
+## 2026-07-14 — price_monitor DB写锁死锁根治（跟进修复）
+
+### 发现问题
+上次诊断只做了临时修复（kill D状态进程 + WAL checkpoint），Dad要求直接根治。
+
+### 修改了什么（三副本同步）
+- **price_monitor.py** (profile/scripts): 统一 BEGIN IMMEDIATE 包裹所有写操作，替代调用 write_holdings_batch/write_portfolio_summary/write_live_prices。5次重试+指数退避(1s→2s→4s→8s→16s)。重试耗尽后自动 emergency WAL checkpoint。try/except 确保 conn 始终释放。
+- **MoFin/scripts/price_monitor.py** — 同步修复
+- **MoFin/price_monitor.py** — 同步修复（保留其 cash_log 优先的现金读取逻辑）
+- **CHANGELOG.md** — 新增2026-07-14条目
+
+### 文件
+- Modified: /home/hmo/.hermes/profiles/position-analyst/scripts/price_monitor.py
+- Modified: /home/hmo/MoFin/scripts/price_monitor.py
+- Modified: /home/hmo/MoFin/price_monitor.py
+- Modified: /home/hmo/MoFin/CHANGELOG.md
@@ -23,7 +23,7 @@ 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, write_holdings_batch, write_portfolio_summary, write_live_prices
+    from mofin_db import get_conn, DB_PATH
    from mo_models import calc_total_mv, calc_total_assets
    from mo_data import read_decisions
    HAS_DB = True
@@ -125,13 +125,32 @@ def refresh_data_prices():
    prices = fetch_all_prices(list(all_codes))
    updated = len(prices)

-    # === 同步实时价到 mofin.db（带重试防锁） ===
+    # === 弹性同步实时价到 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:
-        for db_attempt in range(3):
+        # 先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 更新数据
+                # ── 构建 holdings 更新数据 ──
                db_holdings = []
                for r in conn.execute("SELECT * FROM holdings WHERE is_active=1"):
                    h = dict(r)
@@ -143,20 +162,28 @@ def refresh_data_prices():
                            h['change_pct'] = float(change_pct) if change_pct else 0
                    db_holdings.append(h)

-                # 写入DB持仓价格（write_holdings_batch 用 ON CONFLICT UPDATE 只改价格字段）
-                ok, msg = write_holdings_batch(conn, db_holdings)
-                if not ok:
-                    conn.close()
-                    if db_attempt < 2:
-                        wait = (db_attempt + 1) * 2
-                        print(f"⏳ DB写持仓失败: {msg} → {wait}s后重试", file=sys.stderr)
-                        time.sleep(wait)
-                        continue
-                    else:
-                        print(f"❌ DB写持仓失败（3次重试耗尽）: {msg}", file=sys.stderr)
-                        break
+                # ── 写 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'),
+                    ))

-                # 重新计算市值（现金从cash_log取最新Dad确认值——price_monitor不再持有现金权威）
+                # ── 写 portfolio_summary ──
                mv = calc_total_mv(db_holdings)
                # 从cash_log读取最新verified现金（Dad确认的才是权威），不读portfolio_summary
                latest = conn.execute(
@@ -167,7 +194,6 @@ def refresh_data_prices():
                    db_cash = latest['cash_after'] or 0.0
                    db_frozen = latest['frozen_after'] or 0.0
                else:
-                    # 首次运行/无cash_log记录时回退到summary现存值
                    existing = conn.execute(
                        'SELECT cash, frozen_cash FROM portfolio_summary WHERE id=1'
                    ).fetchone()
@@ -175,36 +201,76 @@ def refresh_data_prices():
                    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
-                write_portfolio_summary(conn, {
-                    'total_mv': mv,
-                    'total_assets': assets,
-                    'stock_value': mv,
-                    'cash': db_cash,
-                    'frozen_cash': db_frozen,
-                    'position_pct': position_pct,
-                    'currency': 'CNY',
-                })
-                # 写实时价格表（供 read_live_prices 消费）
-                live = {h['code']: {'price': h.get('price',0), 'change_pct': h.get('change_pct',0)}
-                        for h in db_holdings if h.get('code')}
-                write_live_prices(conn, live)
+                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)
+                        )
+
                conn.commit()
                conn.close()
+                conn = None
                if db_attempt > 0:
                    print(f"DB同步成功（第{db_attempt+1}次重试）")
                break  # success
-            except sqlite3.OperationalError as e:
-                conn.close()
-                if db_attempt < 2:
-                    wait = (db_attempt + 1) * 2
-                    print(f"⏳ DB锁等待（第{db_attempt+1}次）: {e} → {wait}s后重试", file=sys.stderr)
-                    time.sleep(wait)
+
+            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同步失败（3次重试耗尽）: {e}", file=sys.stderr)
+                    print(f"❌ DB错误: {e}", file=sys.stderr)
+                    break
            except Exception as e:
-                conn.close()
+                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:
+            print("❌ DB同步失败（所有重试耗尽）", file=sys.stderr)
+            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

@@ -17,7 +17,7 @@ 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, write_holdings_batch, write_portfolio_summary, write_live_prices
+    from mofin_db import get_conn, DB_PATH
    from mo_models import calc_total_mv, calc_total_assets
    HAS_DB = True
 except ImportError:
@@ -139,13 +139,32 @@ def refresh_data_prices():
    prices = fetch_all_prices(list(all_codes))
    updated = len(prices)

-    # === 同步实时价到 mofin.db（带重试防锁） ===
+    # === 弹性同步实时价到 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:
-        for db_attempt in range(3):
+        # 先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 更新数据
+                # ── 构建 holdings 更新数据 ──
                db_holdings = []
                for r in conn.execute("SELECT * FROM holdings WHERE is_active=1"):
                    h = dict(r)
@@ -157,22 +176,29 @@ def refresh_data_prices():
                            h['change_pct'] = float(change_pct) if change_pct else 0
                    db_holdings.append(h)

-                # 写入DB持仓价格（write_holdings_batch 用 ON CONFLICT UPDATE 只改价格字段）
-                ok, msg = write_holdings_batch(conn, db_holdings)
-                if not ok:
-                    conn.close()
-                    if db_attempt < 2:
-                        wait = (db_attempt + 1) * 2
-                        print(f"⏳ DB写持仓失败: {msg} → {wait}s后重试", file=sys.stderr)
-                        time.sleep(wait)
-                        continue
-                    else:
-                        print(f"❌ DB写持仓失败（3次重试耗尽）: {msg}", file=sys.stderr)
-                        break
+                # ── 写 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'),
+                    ))

-                # 重新计算市值（不变现金——DB的cash是权威）
+                # ── 写 portfolio_summary ──
                mv = calc_total_mv(db_holdings)
-                # 读取DB当前的现金和冻结，不覆盖
                existing = conn.execute(
                    'SELECT cash, frozen_cash FROM portfolio_summary WHERE id=1'
                ).fetchone()
@@ -180,41 +206,88 @@ def refresh_data_prices():
                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
-                write_portfolio_summary(conn, {
-                    'total_mv': mv,
-                    'total_assets': assets,
-                    'stock_value': mv,
-                    'cash': db_cash,
-                    'frozen_cash': db_frozen,
-                    'position_pct': position_pct,
-                    'currency': 'CNY',
-                })
-                # 写实时价格表（供 read_live_prices 消费）
-                live = {h['code']: {'price': h.get('price',0), 'change_pct': h.get('change_pct',0)}
-                        for h in db_holdings if h.get('code')}
-                # 补充自选股/策略股的价格（它们不在holdings表中）
+                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 live:
-                        live[code] = {'price': pdata[0] if isinstance(pdata, (list, tuple)) else pdata.get('price',0),
-                                      'change_pct': pdata[1] if isinstance(pdata, (list, tuple)) else pdata.get('change_pct',0)}
-                write_live_prices(conn, live)
+                    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 as e:
-                conn.close()
-                if db_attempt < 2:
-                    wait = (db_attempt + 1) * 2
-                    print(f"⏳ DB锁等待（第{db_attempt+1}次）: {e} → {wait}s后重试", file=sys.stderr)
-                    time.sleep(wait)
+
+            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同步失败（3次重试耗尽）: {e}", file=sys.stderr)
+                    print(f"❌ DB错误: {e}", file=sys.stderr)
+                    break
            except Exception as e:
-                conn.close()
+                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