feat: DB-first architecture with lock-safe writes

- price_monitor: writes live prices to both JSON and mofin.db (holdings + live_prices + portfolio_summary)
- mofin_db: added execute_with_retry/commit_with_retry with exponential backoff on 'database is locked'
- mofin_db: increased timeout 5s->15s, added PRAGMA busy_timeout=15000
- price_monitor retry loop: fixed break-before-if-ok bug (was not retrying on write failure)
- DB connection: WAL mode + retry decorator for all write operations
- cash sync: preserves DB authoritative cash (JSON cash not pushed to DB)

This is the DB-first version. JSON writes remain for dashboard compatibility.
Next step: remove JSON writes entirely for full DB-only architecture.

scripts/check_db_paths.py

@@ -0,0 +1,22 @@
+import sqlite3
+
+# Script connects to THIS db
+proj_db = '/home/hmo/projects/MoFin/data/mofin.db'
+# Real data lives in THIS db
+real_db = '/home/hmo/web-dashboard/data/mofin.db'
+
+for label, path in [("project", proj_db), ("real", real_db)]:
+    db = sqlite3.connect(path)
+    tables = [r[0] for r in db.execute("SELECT name FROM sqlite_master WHERE type='table'")]
+    print(f"{label} db ({path}): {len(tables)} tables")
+    for t in tables:
+        if t == 'todos':
+            sql = db.execute(f"SELECT sql FROM sqlite_master WHERE name='{t}'").fetchone()
+            print(f"  {t}: {sql[0][:100] if sql else 'no sql'}")
+        elif t in ('holdings', 'holding_strategies', 'watchlist_stocks', 'portfolio_summary'):
+            cnt = db.execute(f"SELECT COUNT(*) FROM {t}").fetchone()[0]
+            print(f"  {t}: {cnt} rows")
+        else:
+            cnt = db.execute(f"SELECT COUNT(*) FROM {t}").fetchone()[0]
+            print(f"  {t}: {cnt} rows")
+    db.close()

scripts/check_todos_schema.py

@@ -0,0 +1,8 @@
+import sqlite3
+
+for label, path in [("project", '/home/hmo/projects/MoFin/data/mofin.db'), ("real", '/home/hmo/web-dashboard/data/mofin.db')]:
+    db = sqlite3.connect(path)
+    sql = db.execute("SELECT sql FROM sqlite_master WHERE name='todos'").fetchone()
+    print(f"=== {label}: {path} ===")
+    print(sql[0] if sql else "NOT FOUND")
+    db.close()

scripts/fix_self_todo_path.py

@@ -0,0 +1,6 @@
+"""Fix DB_PATH in self_todo_executor.py"""
+path = '/home/hmo/.hermes/profiles/position-analyst/scripts/self_todo_executor.py'
+content = open(path).read()
+content = content.replace('projects/MoFin/data', 'web-dashboard/data')
+open(path, 'w').write(content)
+print("DB_PATH fixed to web-dashboard/data/mofin.db")

scripts/fix_todos_db.py

@@ -0,0 +1,47 @@
+"""Fix: unify todos table schema across project and real DB"""
+import sqlite3
+
+project_db = '/home/hmo/projects/MoFin/data/mofin.db'
+real_db = '/home/hmo/web-dashboard/data/mofin.db'
+
+# Zhiwei's canonical schema (from project db)
+target_schema = """
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    title TEXT NOT NULL,
+    description TEXT,
+    status TEXT DEFAULT 'pending',
+    priority TEXT DEFAULT 'medium',
+    source TEXT DEFAULT 'manual',
+    fix_action TEXT,
+    retry_count INTEGER DEFAULT 0,
+    note TEXT,
+    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+"""
+
+def ensure_todos(db_path, label):
+    db = sqlite3.connect(db_path)
+    existing = db.execute("SELECT name FROM sqlite_master WHERE name='todos'").fetchone()
+    if not existing:
+        db.execute(f"CREATE TABLE todos ({target_schema})")
+        print(f"{label}: created todos table")
+    else:
+        # Ensure all columns exist
+        existing_cols = {r[1] for r in db.execute("PRAGMA table_info(todos)")}
+        needed = {'title', 'description', 'status', 'priority', 'source', 'fix_action',
+                  'retry_count', 'note', 'created_at', 'updated_at'}
+        missing = needed - existing_cols
+        for col in missing:
+            if col in ('retry_count',):
+                db.execute(f"ALTER TABLE todos ADD COLUMN {col} INTEGER DEFAULT 0")
+            elif col in ('created_at', 'updated_at'):
+                db.execute(f"ALTER TABLE todos ADD COLUMN {col} TIMESTAMP DEFAULT CURRENT_TIMESTAMP")
+            else:
+                db.execute(f"ALTER TABLE todos ADD COLUMN {col} TEXT")
+        print(f"{label}: checked, {len(missing)} missing columns added" if missing else f"{label}: schema OK")
+    db.commit()
+    db.close()
+
+ensure_todos(project_db, "project db")
+ensure_todos(real_db, "real db")
+print("\nDone. Both DBs now have matching todos schema.")

scripts/stale_push_wlin.py

@@ -20,6 +20,7 @@ import threading
 import time
 from datetime import datetime, time
 from mo_data import read_portfolio, read_decisions
+from mofin_db import get_conn
 
 # ── MoFin unified model ──────────────────────────────────────────────
 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
@@ -63,9 +64,7 @@ def fetch_trend_data(code):
     # 价格从 DB 读取，不再自拉腾讯 API
     current = 0
     try:
-        import sqlite3
-        db = sqlite3.connect('/home/hmo/web-dashboard/data/mofin.db')
-        db.row_factory = sqlite3.Row
+        db = get_conn()
         row = db.execute("SELECT price FROM holdings WHERE code=? AND is_active=1", (code,)).fetchone()
         if not row:
             row = db.execute("SELECT price FROM watchlist_stocks WHERE code=? AND is_active=1", (code,)).fetchone()
@@ -155,8 +154,7 @@ def load_macro_line():
     parts = []
     try:
         # 优先 DB
-        import sqlite3
-        db = sqlite3.connect("/home/hmo/MoFin/data/mofin.db")
+        db = get_conn()
         row = db.execute(
             "SELECT structure FROM macro_context_log "
             "WHERE has_valid_data=1 ORDER BY created_at DESC LIMIT 1"

scripts/verify_self_todo.py

@@ -0,0 +1,32 @@
+"""Verify self_todo_executor works with real DB"""
+import subprocess
+script = '/home/hmo/.hermes/profiles/position-analyst/scripts/self_todo_executor.py'
+
+# Test 1: DB_PATH
+content = open(script).read()
+if 'web-dashboard/data/mofin.db' in content:
+    print("DB_PATH: OK")
+else:
+    print("DB_PATH: WRONG")
+    exit(1)
+
+# Test 2: script can import and run
+try:
+    import importlib.util
+    spec = importlib.util.spec_from_file_location("executor", script)
+    mod = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)
+    print("Import: OK")
+except Exception as e:
+    print(f"Import: FAIL -> {e}")
+    exit(1)
+
+# Test 3: get_pending works
+try:
+    rows = mod.get_pending()
+    print(f"get_pending: OK ({len(rows)} pending)")
+except Exception as e:
+    print(f"get_pending: FAIL -> {e}")
+    exit(1)
+
+print("\nAll checks passed.")