feat: mo_data.py unified read layer (DB-first, JSON fallback) + cash_log table + batch JSON→DB migration (16 files)

mo_data.py

@@ -0,0 +1,198 @@
+#!/usr/bin/env python3
+"""
+mo_data.py — MoFin 统一数据读取层
+
+替代 json.load(open(portfolio.json)) 等直接读 JSON 文件的方式。
+所有数据从 DB 读取，币种字段强制存在。
+
+用法:
+  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, os
+from datetime import datetime
+
+DB_PATH = '/home/hmo/web-dashboard/data/mofin.db'
+
+# JSON 文件路径（冷备，仅当 DB 不可用时 fallback）
+PORTFOLIO_JSON = '/home/hmo/web-dashboard/data/portfolio.json'
+DECISIONS_JSON = '/home/hmo/web-dashboard/data/decisions.json'
+WATCHLIST_JSON = '/home/hmo/web-dashboard/data/watchlist.json'
+
+
+def _get_db():
+    db = sqlite3.connect(DB_PATH)
+    db.row_factory = sqlite3.Row
+    return db
+
+
+# ── portfolio.json → holdings + portfolio_summary ─────────────────
+
+def read_portfolio():
+    """返回 portfolio.json 等价 dict。DB 优先，JSON 冷备。"""
+    try:
+        db = _get_db()
+        # holdings
+        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')  # 兼容旧代码的 _currency 字段名
+            holdings.append(h)
+        
+        # summary
+        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", ""),
+        }
+    except Exception:
+        pass
+    
+    # JSON 冷备
+    try:
+        return json.load(open(PORTFOLIO_JSON, encoding='utf-8'))
+    except:
+        return {"holdings": [], "total_assets": 0, "cash": 0, "frozen_cash": 0}
+
+
+# ── decisions.json → holding_strategies ────────────────────────────
+
+def read_decisions():
+    """返回 decisions.json 等价 dict。DB 优先，JSON 冷备。"""
+    try:
+        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 "
+            "FROM holding_strategies WHERE status IN ('active','updated') "
+            "ORDER BY code"
+        ).fetchall()
+        
+        decisions = []
+        for r in rows:
+            d = dict(r)
+            # 还原 trigger 字段
+            if d.get('trigger_json'):
+                try:
+                    d['trigger'] = json.loads(d['trigger_json'])
+                except:
+                    d['trigger'] = {}
+            else:
+                d['trigger'] = {}
+            # 还原 changelog 字段
+            if d.get('changelog_json'):
+                try:
+                    d['changelog'] = json.loads(d['changelog_json'])
+                except:
+                    d['changelog'] = []
+            else:
+                d['changelog'] = []
+            decisions.append(d)
+        
+        db.close()
+        
+        return {
+            "decisions": decisions,
+            "total": len(decisions),
+            "regenerated_at": datetime.now().strftime('%Y-%m-%d %H:%M'),
+        }
+    except Exception:
+        pass
+    
+    try:
+        return json.load(open(DECISIONS_JSON, encoding='utf-8'))
+    except:
+        return {"decisions": [], "total": 0}
+
+
+# ── watchlist.json → watchlist_stocks ──────────────────────────────
+
+def read_watchlist():
+    """返回 watchlist.json 等价 dict。DB 优先，JSON 冷备。"""
+    try:
+        db = _get_db()
+        rows = db.execute(
+            "SELECT code, name, price, entry_low, entry_high, "
+            "stop_loss, currency, source, source_detail, notes, "
+            "added_by, added_at, analysis_json "
+            "FROM watchlist_stocks WHERE is_active=1"
+        ).fetchall()
+        
+        stocks = []
+        for r in rows:
+            s = dict(r)
+            if s.get('source_detail'):
+                try: s['source_detail'] = json.loads(s['source_detail'])
+                except: pass
+            if s.get('analysis_json'):
+                try: s['analysis'] = json.loads(s['analysis_json'])
+                except: s['analysis'] = {}
+            else:
+                s['analysis'] = {}
+            stocks.append(s)
+        
+        db.close()
+        
+        return {
+            "stocks": stocks,
+            "updated_at": datetime.now().strftime('%Y-%m-%d %H:%M'),
+        }
+    except Exception:
+        pass
+    
+    try:
+        return json.load(open(WATCHLIST_JSON, encoding='utf-8'))
+    except:
+        return {"stocks": []}
+
+
+# ── 便捷函数 ────────────────────────────────────────────────────────
+
+def read_portfolio_json():
+    """别名（兼容旧代码直接 import 后调用）"""
+    return read_portfolio()
+
+def read_decisions_json():
+    """别名"""
+    return read_decisions()
+
+def read_watchlist_json():
+    """别名"""
+    return read_watchlist()
+
+
+# ── 自检 ────────────────────────────────────────────────────────────
+
+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")