diff --git a/deploy/profile-scripts/sync_profile_scripts.sh b/deploy/profile-scripts/sync_profile_scripts.sh
index 96d81d44..5d8d3349 100755
--- a/deploy/profile-scripts/sync_profile_scripts.sh
+++ b/deploy/profile-scripts/sync_profile_scripts.sh
@@ -11,6 +11,19 @@ for f in "$SRC"/*.py; do
     count=$((count+1))
 done
 ln -f "$SRC/sync_profile_scripts.sh" "$DST/sync_profile_scripts.sh" 2>/dev/null || true
+
+# ── 库文件 SSOT：canonical 在 MoFin 根目录，其他位置只许硬链（红线#6）──
+# git checkout/merge 会重写文件破坏硬链，这里强制恢复
+ROOT="/home/hmo/MoFin"
+for lib in mofin_db.py mo_data.py; do
+    if [ -f "$ROOT/$lib" ]; then
+        ln -f "$ROOT/$lib" "$ROOT/deploy/profile-scripts/$lib" 2>/dev/null || true
+        ln -f "$ROOT/$lib" "$ROOT/scripts/$lib" 2>/dev/null || true
+        ln -f "$ROOT/$lib" "/home/hmo/web-dashboard/$lib" 2>/dev/null || true
+        ln -f "$ROOT/$lib" "$DST/$lib" 2>/dev/null || true
+    fi
+done
+
 echo "synced $count scripts (hardlink)"
 mkdir -p /home/hmo/MoFin/gateway/logs
 echo "$(date '+%F %T') synced $count scripts" >> /home/hmo/MoFin/gateway/logs/link_sync.log
\ No newline at end of file
diff --git a/mo_data.py b/mo_data.py
new file mode 100644
index 00000000..cb168a13
--- /dev/null
+++ b/mo_data.py
@@ -0,0 +1,316 @@
+#!/usr/bin/env python3
+"""
+mo_data.py — MoFin 统一数据层（纯 DB）
+
+所有数据从 SQLite 读取。不做 JSON fallback。
+JSON 文件已弃用，仅保留为历史备份。
+
+用法:
+  from mo_data import read_portfolio, read_decisions, read_watchlist
+
+  pf = read_portfolio()   # 返回和 portfolio.json 一样的 dict 结构
+  dec = read_decisions()  # 返回和 decisions.json 一样的 dict 结构
+  wl = read_watchlist()   # 返回和 watchlist.json 一样的 dict 结构
+"""
+
+import sqlite3, json, sys
+from datetime import datetime
+from pathlib import Path
+
+DB_PATH = '/home/hmo/MoFin/data/mofin.db'
+SCRIPT_DIR = Path('/home/hmo/MoFin/scripts')
+
+
+def _get_db():
+    db = sqlite3.connect(DB_PATH)
+    db.row_factory = sqlite3.Row
+    return db
+
+
+# ── portfolio ─────────────────────────────────────────────────────
+
+def read_portfolio():
+    """返回 portfolio.json 等价 dict。纯 DB。"""
+    db = _get_db()
+    rows = db.execute(
+        "SELECT code, name, shares, cost, price, market_value, "
+        "change_pct, currency, position_pct "
+        "FROM holdings WHERE is_active=1"
+    ).fetchall()
+    holdings = []
+    for r in rows:
+        h = dict(r)
+        h['_currency'] = h.get('currency', 'CNY')
+        holdings.append(h)
+
+    sum_row = db.execute("SELECT * FROM portfolio_summary WHERE id=1").fetchone()
+    summary = dict(sum_row) if sum_row else {}
+
+    db.close()
+
+    return {
+        "holdings": holdings,
+        "total_assets": summary.get("total_assets", 0),
+        "total_mv": summary.get("total_mv", 0),
+        "stock_value": summary.get("stock_value", summary.get("total_mv", 0)),
+        "cash": summary.get("cash", 0),
+        "frozen_cash": summary.get("frozen_cash", 0),
+        "position_pct": summary.get("position_pct", 0),
+        "currency": summary.get("currency", "CNY"),
+        "updated_at": summary.get("updated_at", ""),
+    }
+
+
+# ── decisions ─────────────────────────────────────────────────────
+
+def _parse_json(val, default):
+    if val:
+        try: return json.loads(val)
+        except: pass
+    return default
+
+
+def read_decisions():
+    """返回 decisions.json 等价 dict。纯 DB。"""
+    db = _get_db()
+    rows = db.execute(
+        "SELECT code, name, version, price, cost, shares, "
+        "stop_loss, take_profit, entry_low, entry_high, "
+        "currency, strategy_type, action, timing_signal, "
+        "rr_ratio, tech_snapshot, stock_category, sector_context, "
+        "status, trigger_json, changelog_json, source, reason, "
+        "created_at, updated_at, "
+        "avg_price, decision_timestamp, note, quality_check, "
+        "quality_checked_at, quality_issues_json, position_advice, "
+        "signal_factors_json, time_horizon, decision_type, tag "
+        "FROM holding_strategies WHERE status IN ('active','updated') "
+        "ORDER BY code"
+    ).fetchall()
+
+    decisions = []
+    for r in rows:
+        d = dict(r)
+        d['trigger'] = _parse_json(r['trigger_json'], {})
+        d['changelog'] = _parse_json(r['changelog_json'], [])
+        d['quality_issues'] = _parse_json(r['quality_issues_json'], {})
+        d['signal_factors'] = _parse_json(r['signal_factors_json'], [])
+        d['timestamp'] = r['decision_timestamp'] or r['created_at'] or ''
+        d['type'] = r['decision_type'] or r['strategy_type'] or '持仓策略'
+        decisions.append(d)
+
+    db.close()
+
+    return {
+        "decisions": decisions,
+        "total": len(decisions),
+        "regenerated_at": datetime.now().strftime('%Y-%m-%d %H:%M'),
+    }
+
+
+# ── watchlist ─────────────────────────────────────────────────────
+
+def read_watchlist():
+    """返回 watchlist 等价 dict。纯 DB。
+    从 holding_strategies（自选策略）读取，watchlist_stocks 已废弃。"""
+    db = _get_db()
+    # 主数据源：holding_strategies 自选策略
+    rows = db.execute(
+        "SELECT code, name, price, entry_low, entry_high, "
+        "stop_loss, currency, updated_at "
+        "FROM holding_strategies WHERE status='active' AND decision_type='自选策略'"
+    ).fetchall()
+    
+    stocks = []
+    seen = set()
+    for r in rows:
+        code = str(r["code"])
+        if code in seen:
+            continue
+        seen.add(code)
+        stocks.append({
+            "code": code,
+            "name": r["name"] or "",
+            "price": r["price"] or 0,
+            "entry_low": r["entry_low"] or 0,
+            "entry_high": r["entry_high"] or 0,
+            "stop_loss": r["stop_loss"] or 0,
+            "currency": r["currency"] or "CNY",
+            "added_at": r["updated_at"] or "",
+            "analysis": {},
+        })
+    
+    return {"stocks": stocks, "total": len(stocks)}
+
+    db.close()
+
+    return {
+        "stocks": stocks,
+        "updated_at": datetime.now().strftime('%Y-%m-%d %H:%M'),
+    }
+
+
+# ── 便捷别名 ───────────────────────────────────────────────────────
+
+def read_portfolio_json():
+    return read_portfolio()
+
+def read_decisions_json():
+    return read_decisions()
+
+def read_watchlist_json():
+    return read_watchlist()
+
+
+# ── 统一价格获取（唯一入口，禁止各脚本自拉API）──
+
+def get_price(code, max_age_minutes=5, use_stale_fallback=True):
+    """获取单只股票最新价格。
+    
+    优先级: live_prices(DB) → stock_quote(API兜底)
+    - live_prices 有且不超过 max_age_minutes → 直接返回
+    - 没有或过期 → 调 stock_quote 拉，写回 live_prices
+    - 都失败 → 返回 (None, None)
+    
+    返回 (price, change_pct)，两值都是 float 或 None。
+    """
+    from mofin_db import get_price_from_db
+    from datetime import datetime, timedelta
+    
+    # 1. 先读 DB
+    try:
+        db_price, db_chg = get_price_from_db(code)
+        if db_price is not None and db_price > 0:
+            # 检查时效性
+            conn = __import__('sqlite3').connect(str(DB_PATH))
+            row = conn.execute(
+                "SELECT updated_at FROM live_prices WHERE code=?",
+                (str(code).strip(),)
+            ).fetchone()
+            conn.close()
+            if row and row[0]:
+                try:
+                    updated = datetime.strptime(row[0], "%Y-%m-%d %H:%M:%S")
+                    age = (datetime.now() - updated).total_seconds() / 60
+                    if age <= max_age_minutes:
+                        return (db_price, db_chg)
+                except:
+                    pass
+            else:
+                return (db_price, db_chg)
+    except Exception:
+        pass
+    
+    # 2. DB 没有或过期 → 调 stock_quote
+    if not use_stale_fallback:
+        return (None, None)
+    
+    try:
+        import subprocess, json
+        r = subprocess.run(
+            [sys.executable, str(SCRIPT_DIR / "stock_quote.py"), str(code)],
+            capture_output=True, text=True, timeout=15
+        )
+        if r.returncode == 0:
+            data = json.loads(r.stdout.strip())
+            price = float(data.get("price", 0))
+            chg = float(data.get("change_pct", 0))
+            if price > 0:
+                # 写回 live_prices
+                try:
+                    conn = __import__('sqlite3').connect(str(DB_PATH))
+                    conn.execute("""
+                        INSERT OR REPLACE INTO live_prices (code, price, change_pct, updated_at)
+                        VALUES (?, ?, ?, datetime('now','localtime'))
+                    """, (str(code).strip(), price, chg))
+                    conn.commit()
+                    conn.close()
+                except:
+                    pass
+                return (price, chg)
+    except Exception:
+        pass
+    
+    return (None, None)
+
+
+def get_prices_batch(codes, max_age_minutes=5):
+    """批量获取价格，返回 {code: (price, change_pct)}"""
+    from mofin_db import get_prices_batch_from_db
+    
+    result = {}
+    need_api = []
+    
+    # 1. 批量读 DB
+    try:
+        db_prices = get_prices_batch_from_db(codes)
+        for code in codes:
+            cs = str(code).strip()
+            if cs in db_prices:
+                p, c = db_prices[cs]
+                if p and p > 0:
+                    result[cs] = (p, c)
+                    continue
+            need_api.append(cs)
+    except:
+        need_api = [str(c).strip() for c in codes]
+    
+    # 2. 缺失的调 API
+    if need_api:
+        try:
+            import subprocess, json
+            r = subprocess.run(
+                [sys.executable, str(SCRIPT_DIR / "stock_quote.py")] + need_api,
+                capture_output=True, text=True, timeout=30
+            )
+            if r.returncode == 0:
+                for line in r.stdout.strip().split("\n"):
+                    if not line:
+                        continue
+                    try:
+                        data = json.loads(line)
+                        code = str(data.get("code", "")).strip()
+                        price = float(data.get("price", 0))
+                        chg = float(data.get("change_pct", 0))
+                        if code and price > 0:
+                            result[code] = (price, chg)
+                    except:
+                        pass
+        except:
+            pass
+    
+    return result
+
+
+# ── cash_log 写入 ──────────────────────────────────────────────────
+
+def write_cash_log(cash_before, cash_after, frozen_before, frozen_after,
+                   source, note, verified=0):
+    """记录现金变更到 cash_log 表。"""
+    change_amount = round(cash_after - cash_before, 2) if cash_after is not None and cash_before is not None else 0
+    db = sqlite3.connect(DB_PATH)
+    try:
+        cur = db.execute(
+            """INSERT INTO cash_log
+               (timestamp, cash_before, cash_after, frozen_before, frozen_after,
+                change_amount, source, note, verified)
+               VALUES (datetime('now','localtime'), ?, ?, ?, ?, ?, ?, ?, ?)""",
+            (cash_before, cash_after, frozen_before, frozen_after,
+             change_amount, source, note, verified)
+        )
+        db.commit()
+        return cur.lastrowid
+    finally:
+        db.close()
+
+
+# ── 自检 ───────────────────────────────────────────────────────────
+
+if __name__ == "__main__":
+    pf = read_portfolio()
+    print(f"portfolio: {len(pf.get('holdings',[]))} holdings, total_assets={pf.get('total_assets',0)}")
+
+    dec = read_decisions()
+    print(f"decisions: {len(dec.get('decisions',[]))} entries")
+
+    wl = read_watchlist()
+    print(f"watchlist: {len(wl.get('stocks',[]))} stocks")