#!/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 from datetime import datetime DB_PATH = '/home/hmo/web-dashboard/data/mofin.db' def _get_db(): """获取数据库连接（WAL 模式，15秒超时防并发锁）""" db = sqlite3.connect(DB_PATH, timeout=15) db.row_factory = sqlite3.Row db.execute("PRAGMA journal_mode=WAL") db.execute("PRAGMA busy_timeout=15000") return db # ── portfolio ───────────────────────────────────────────────────── def read_portfolio(): """返回 portfolio.json 等价 dict。纯 DB。 总市值从 holdings 实时计算（shares × price，港股 × 汇率）， 不信任 portfolio_summary 的存储值，因为可能未及时更新。 """ 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() # ── 总市值从 holdings 实时计算 ──────────────────── cash = float(summary.get('cash', 0) or 0) frozen = float(summary.get('frozen_cash', 0) or 0) # 获取港股汇率 import subprocess, os as _os rate = 0.865 # 默认值 try: hk_rate_py = _os.path.join(_os.path.dirname(_os.path.dirname(__file__)), 'hk_rate.py') if _os.path.exists(hk_rate_py): r = subprocess.run( [_os.path.join(_os.path.dirname(_os.path.dirname(__file__)), 'venv', 'bin', 'python3'), hk_rate_py, '--rate'], capture_output=True, text=True, timeout=10 ) if r.returncode == 0 and r.stdout.strip(): rate = float(r.stdout.strip()) except Exception: pass total_mv = 0.0 for h in holdings: p = float(h.get('price', 0) or 0) s = float(h.get('shares', 0) or 0) if h.get('currency') == 'HKD' or (len(str(h.get('code',''))) == 5 and str(h.get('code',''))[0] in ('0','1')): total_mv += s * p * rate else: total_mv += s * p total_mv = round(total_mv, 2) total_assets = round(total_mv + cash + frozen, 2) position_pct = round(total_mv / total_assets * 100, 2) if total_assets > 0 else 0 return { "holdings": holdings, "total_assets": total_assets, "total_mv": total_mv, "stock_value": total_mv, "cash": cash, "frozen_cash": frozen, "position_pct": position_pct, "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 " "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.json 等价 dict。纯 DB。""" 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) s['source_detail'] = _parse_json(r['source_detail'], None) if r['analysis_json']: s['analysis'] = _parse_json(r['analysis_json'], {}) else: s['analysis'] = {} stocks.append(s) 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() # ── 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")