#!/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(): 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 " "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")