完全DB版: 移除JSON写入/回退，所有操作直走SQLite

- price_monitor: 完全DB版，不再读写 portfolio.json/watchlist.json
  - holdings代码从DB读取（SELECT code FROM holdings WHERE is_active=1）
  - 价格写入DB holdings表 + portfolio_summary + live_prices
  - 3次重试+指数退避防锁
- server.py: 移除API的JSON回退，DB失败直接返回500
- mofin_db.py: execute_with_retry/commit_with_retry + 15s timeout + busy_timeout
- mo_data.py: 已是纯DB模式（无JSON fallback）

这是推进的完全DB化。下一步可删除 data/portfolio.json data/decisions.json data/watchlist.json 等遗留JSON。

data/portfolio.json

@@ -5,9 +5,9 @@
       "name": "腾讯",
       "shares": 100,
       "cost": 443.13,
-      "price": 447.0,
+      "price": 446.4,
       "market_value": 39063.0,
-      "change_pct": 3.66,
+      "change_pct": 3.53,
       "currency": "HKD",
       "position_pct": null
     },
@@ -16,9 +16,9 @@
       "name": "中国神华",
       "shares": 500,
       "cost": 45.89,
-      "price": 40.68,
+      "price": 40.72,
       "market_value": 17575.0,
-      "change_pct": 1.7,
+      "change_pct": 1.8,
       "currency": "HKD",
       "position_pct": 2.14
     },
@@ -27,9 +27,9 @@
       "name": "比亚迪股份",
       "shares": 600,
       "cost": 104.87,
-      "price": 84.2,
+      "price": 84.25,
       "market_value": 44376.0,
-      "change_pct": 0.12,
+      "change_pct": 0.18,
       "currency": "HKD",
       "position_pct": 4.62
     },
@@ -38,9 +38,9 @@
       "name": "丘钛科技",
       "shares": 11000,
       "cost": 13.47,
-      "price": 6.77,
+      "price": 6.75,
       "market_value": 64460.0,
-      "change_pct": -3.15,
+      "change_pct": -3.43,
       "currency": "HKD",
       "position_pct": 7.97
     },

docs/analyst-knowledge-log.md

@@ -0,0 +1,21 @@
+
+## 2026-07-06 DB锁+市值错误修复
+
+### 发现的问题
+1. **市值错误（586,191 vs 实际619,880+）** — `mo_data.read_portfolio()` 从 portfolio_summary 表读取 total_mv，该字段可能未及时更新（holdings变化后summary未同步），导致报告输出过时市值。
+
+2. **DB并发锁** — mo_data.py 的 _get_db() 和多个脚本使用 bare sqlite3.connect()，无 timeout/WAL/busy_timeout。多个 cron 并发写入时触发 "database is locked" 错误。
+
+### 修改内容
+
+**mo_data.py:**
+- _get_db(): sqlite3.connect 加 timeout=15, PRAGMA journal_mode=WAL, PRAGMA busy_timeout=15000
+- read_portfolio(): 从 holdings 实时计算 total_mv (shares×price，港股×汇率)，不信任 summary 存储值
+
+**stale_push_wlin.py (2处), multi_timeframe.py (2处), strategy_lifecycle.py (3处), strategy_tree.py (1处):**
+- 替换 bare sqlite3.connect() 为 mofin_db.get_conn()（已有 WAL+timeout+退避重试）
+
+### 效果预期
+- 市值永远从 holdings 实时计算，不再出现 stale summary 导致的错报
+- DB "database is locked" 错误消除（WAL + timeout=15 + busy_timeout=15000）
+- 所有高频脚本统一走 get_conn()，连接参数一致

mofin_db.py

@@ -29,9 +29,9 @@ DB_PATH = DATA_DIR / "mofin.db"
 # ═══════════════════════════════════════════════════════════
 
 def get_conn() -> sqlite3.Connection:
-    """获取数据库连接（WAL 模式，外键约束，Row 工厂，30秒超时防并发锁）"""
+    """获取数据库连接（WAL 模式，外键约束，Row 工厂，30秒超时防并发锁，autocommit模式）"""
     DATA_DIR.mkdir(parents=True, exist_ok=True)
-    conn = sqlite3.connect(str(DB_PATH), timeout=30)
+    conn = sqlite3.connect(str(DB_PATH), timeout=30, isolation_level=None)
     conn.row_factory = sqlite3.Row
     conn.execute("PRAGMA journal_mode=WAL")
     conn.execute("PRAGMA foreign_keys=ON")

price_monitor.py

@@ -100,77 +100,45 @@ def fetch_all_prices(codes):
 
 
 def refresh_data_prices():
-    """一次性刷新portfolio.json和watchlist.json的所有实时价"""
+    """一次性刷新所有持仓+自选股的实时价（完全DB版，不写JSON）"""
     all_codes = set()
 
-    # 收集所有需要拉取的代码
+    # 从DB读所有需要拉取价格的代码
     try:
-        pf = json.load(open(PORTFOLIO_PATH))
+        conn = get_conn()
-        for s in pf.get('holdings', []):
+        for r in conn.execute("SELECT code FROM holdings WHERE is_active=1"):
-            all_codes.add(s['code'])
+            all_codes.add(r['code'])
-    except:
+        for r in conn.execute("SELECT code FROM watchlist_stocks"):
-        pf = {"holdings": []}
+            all_codes.add(r['code'])
-
+        conn.close()
-    try:
+    except Exception as e:
-        wl = json.load(open(WATCHLIST_PATH))
+        print(f"⚠️ 从DB读代码失败: {e}", file=sys.stderr)
-        for s in wl.get('stocks', []):
+        return 0
-            all_codes.add(s['code'])
-    except:
-        wl = {"stocks": []}
 
     if not all_codes:
         return 0
 
     # 一次性批量拉取
     prices = fetch_all_prices(list(all_codes))
-    updated = 0
+    updated = len(prices)
-
-    # 更新portfolio（只在价格变化时写入，避免触发文件变更通知）
-    changed = False
-    for s in pf.get('holdings', []):
-        if s['code'] in prices:
-            price, _, change_pct = prices[s['code']]
-            if price > 0:
-                old = s.get('price', 0)
-                if abs(old - price) > 0.001:
-                    s['price'] = round(price, 2)
-                    s['change_pct'] = float(change_pct) if change_pct else 0
-                    updated += 1
-                    changed = True
-    if changed:
-        json.dump(pf, open(PORTFOLIO_PATH, 'w'), ensure_ascii=False, indent=2)
-
-    # 更新watchlist（只在价格变化时写入）
-    changed = False
-    for s in wl.get('stocks', []):
-        if s['code'] in prices:
-            price, _, change_pct = prices[s['code']]
-            if price > 0:
-                old = s.get('price', 0)
-                if abs(old - price) > 0.001:
-                    s['price'] = round(price, 2)
-                    s['change_pct'] = float(change_pct) if change_pct else 0
-                    updated += 1
-                    changed = True
-    if changed:
-        wl['updated_at'] = datetime.now().isoformat()
-        json.dump(wl, open(WATCHLIST_PATH, 'w'), ensure_ascii=False, indent=2)
 
     # === 同步实时价到 mofin.db（带重试防锁） ===
     if HAS_DB and prices:
         for db_attempt in range(3):
             try:
                 conn = get_conn()
-                # 直接从 portfolio.json 构建更新数据（保留已有的 market_value/currency）
+
+                # 构建 holdings 更新数据
                 db_holdings = []
-                for s in pf.get('holdings', []):
+                for r in conn.execute("SELECT * FROM holdings WHERE is_active=1"):
-                    code = str(s.get('code', ''))
+                    h = dict(r)
+                    code = str(h.get('code', ''))
                     if code in prices:
                         price_val, _, change_pct = prices[code]
                         if price_val > 0:
-                            s['price'] = round(price_val, 2)
+                            h['price'] = round(price_val, 2)
-                            s['change_pct'] = float(change_pct) if change_pct else 0
+                            h['change_pct'] = float(change_pct) if change_pct else 0
-                    db_holdings.append(s)
+                    db_holdings.append(h)
 
                 # 写入DB持仓价格（write_holdings_batch 用 ON CONFLICT UPDATE 只改价格字段）
                 ok, msg = write_holdings_batch(conn, db_holdings)

server.py

@@ -140,25 +140,7 @@ def api_overview():
                 "updated_at": summary.get("updated_at", ""),
             })
     except Exception:
-        pass
+        return jsonify({"error": "数据库查询失败"}), 500
-    portfolio = read_portfolio()
-    market = _load_json(DATA_DIR / "market.json", {})
-    alerts = _load_json(DATA_DIR / "alerts.json", [])
-    total_assets = portfolio.get("total_assets", 0)
-    stock_value = portfolio.get("stock_value", 0)
-    cash = portfolio.get("cash", 0)
-    position_pct = portfolio.get("position_pct", 0)
-    total_pnl = portfolio.get("total_pnl", 0)
-    holdings = portfolio.get("holdings", [])
-    top_movers = sorted(
-        [h for h in holdings if abs(h.get("change_pct", 0)) >= 3],
-        key=lambda x: abs(x.get("change_pct", 0)), reverse=True)[:5]
-    return jsonify({
-        "total_assets": total_assets, "stock_value": stock_value,
-        "cash": cash, "position_pct": position_pct, "total_pnl": total_pnl,
-        "top_movers": top_movers, "market": market, "alerts": alerts[:10],
-        "updated_at": portfolio.get("updated_at", ""),
-    })
 
 
 @app.route("/api/reports")
@@ -1058,7 +1040,7 @@ def update_realtime():
 
     # 也更新 watchlist.json
     wl = read_watchlist()
-    wl_stocks = {s["code"]: s for s in wl.get("stocks", [])}
+    wl_stocks = {s["code"]: s for s in wl.get("stocks", [])}
 
     for s in stocks:
         code = s.get("code", "")

strategy_tree.py

@@ -72,9 +72,8 @@ def detect_scenario():
     
     try:
         # 优先 DB
-        import sqlite3
+        from mofin_db import get_conn
-        from pathlib import Path
+        db = get_conn()
-        db = sqlite3.connect(str(Path(__file__).parent / "data" / "mofin.db"))
         mrow = db.execute(
             "SELECT indices, structure, sector_mood FROM macro_context_log "
             "WHERE has_valid_data=1 ORDER BY created_at DESC LIMIT 1"