diff --git a/__pycache__/multi_timeframe.cpython-312.pyc b/__pycache__/multi_timeframe.cpython-312.pyc index 82ad25e..4735bd8 100644 Binary files a/__pycache__/multi_timeframe.cpython-312.pyc and b/__pycache__/multi_timeframe.cpython-312.pyc differ diff --git a/__pycache__/strategy_lifecycle.cpython-312.pyc b/__pycache__/strategy_lifecycle.cpython-312.pyc index d15b460..c786b29 100644 Binary files a/__pycache__/strategy_lifecycle.cpython-312.pyc and b/__pycache__/strategy_lifecycle.cpython-312.pyc differ diff --git a/data/mofin.db-shm b/data/mofin.db-shm deleted file mode 100644 index 2115847..0000000 Binary files a/data/mofin.db-shm and /dev/null differ diff --git a/data/mofin.db-wal b/data/mofin.db-wal deleted file mode 100644 index 816a4be..0000000 Binary files a/data/mofin.db-wal and /dev/null differ diff --git a/data/portfolio.json b/data/portfolio.json index d511e98..6791eb8 100644 --- a/data/portfolio.json +++ b/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 }, diff --git a/docs/analyst-knowledge-log.md b/docs/analyst-knowledge-log.md new file mode 100644 index 0000000..468fd8c --- /dev/null +++ b/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()，连接参数一致 diff --git a/mofin_db.py b/mofin_db.py index 7f705ec..ca765a4 100644 --- a/mofin_db.py +++ b/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") diff --git a/price_monitor.py b/price_monitor.py index 6c93889..9ced72e 100644 --- a/price_monitor.py +++ b/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)) - for s in pf.get('holdings', []): - all_codes.add(s['code']) - except: - pf = {"holdings": []} - - try: - wl = json.load(open(WATCHLIST_PATH)) - for s in wl.get('stocks', []): - all_codes.add(s['code']) - except: - wl = {"stocks": []} + 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']) + conn.close() + except Exception as e: + print(f"⚠️ 从DB读代码失败: {e}", file=sys.stderr) + return 0 if not all_codes: return 0 # 一次性批量拉取 prices = fetch_all_prices(list(all_codes)) - updated = 0 - - # 更新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) + updated = len(prices) # === 同步实时价到 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', []): - code = str(s.get('code', '')) + 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: - s['price'] = round(price_val, 2) - s['change_pct'] = float(change_pct) if change_pct else 0 - db_holdings.append(s) + h['price'] = round(price_val, 2) + 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) diff --git a/server.py b/server.py index ed73c9d..0b46cbb 100644 --- a/server.py +++ b/server.py @@ -140,25 +140,7 @@ def api_overview(): "updated_at": summary.get("updated_at", ""), }) except Exception: - pass - 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", ""), - }) + return jsonify({"error": "数据库查询失败"}), 500 @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", "") diff --git a/strategy_tree.py b/strategy_tree.py index 3a8edfc..e4300ed 100644 --- a/strategy_tree.py +++ b/strategy_tree.py @@ -72,9 +72,8 @@ def detect_scenario(): try: # 优先 DB - import sqlite3 - from pathlib import Path - db = sqlite3.connect(str(Path(__file__).parent / "data" / "mofin.db")) + from mofin_db import get_conn + db = get_conn() mrow = db.execute( "SELECT indices, structure, sector_mood FROM macro_context_log " "WHERE has_valid_data=1 ORDER BY created_at DESC LIMIT 1"