diff --git a/CHANGELOG.md b/CHANGELOG.md index 91090456..aa0f3c32 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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 diff --git a/docs/analyst-knowledge-log.md b/docs/analyst-knowledge-log.md index 01cddbc7..4285c5d5 100644 --- a/docs/analyst-knowledge-log.md +++ b/docs/analyst-knowledge-log.md @@ -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 diff --git a/price_monitor.py b/price_monitor.py index bd5eb615..25765e67 100644 --- a/price_monitor.py +++ b/price_monitor.py @@ -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 diff --git a/scripts/price_monitor.py b/scripts/price_monitor.py index a2dcb491..802f8442 100644 --- a/scripts/price_monitor.py +++ b/scripts/price_monitor.py @@ -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