refactor: 消费者切 SQLite 优先读取

切换策略: SQLite 优先 → 失败回退 JSON

price_events (100%覆盖):
- strategy_feedback.py: run() 优先 query_price_events()
- system_health_check.py: 优先 query_price_events() + query_price_events_by_date()

stock_sector_map (100%覆盖):
- strategy_lifecycle.py: load_stock_sector_map() 优先 stock_sectors 表

market.json (85%覆盖):
- strategy_lifecycle.py: load_market_context() 优先 query_latest_market()
- market_insight.py: generate() 优先 query_latest_market()

portfolio.json + watchlist.json (70%覆盖):
- strategy_lifecycle.py: regenerate_all() 优先 query_holdings() + query_watchlist()
- server.py: /api/portfolio, /api/watchlist, /api/overview, /api/market 优先 SQLite

所有改动保留 JSON 回退路径，SQLite 不可用时自动降级

market_insight.py

@@ -55,11 +55,32 @@ def load_holding_industry_map():
 
 
 def generate():
-    # 加载数据
+    # 优先从 SQLite 读取市场数据
+    try:
+        from mofin_db import get_conn, query_latest_market
+        conn = get_conn()
+        market = query_latest_market(conn)
+        conn.close()
+        if market and market.get("sectors"):
+            sectors = market["sectors"]
+            top_gainers = market.get("top_gainers", [])
+            top_losers = market.get("top_losers", [])
+            mood = market.get("mood", "unknown")
+            up_ratio = market.get("up_ratio", 0)
+            timestamp = market.get("timestamp", "")
+            # 字段名适配
+            for s in sectors:
+                s["change"] = s.get("change_pct", 0)
+            for g in top_gainers:
+                g["change"] = g.get("change_pct", 0)
+            for l in top_losers:
+                l["change"] = l.get("change_pct", 0)
+        else:
+            raise Exception("no data")
+    except Exception:
         market_path = DATA_DIR / "market.json"
         with open(market_path, "r", encoding="utf-8") as f:
             market = json.load(f)
-
         sectors = market.get("sectors", [])
         top_gainers = market.get("top_gainers", [])
         top_losers = market.get("top_losers", [])

server.py

@@ -33,6 +33,20 @@ def _load_json(path, default=None):
         return {} if default is None else default
 
 
+def _load_from_db(query_func, json_path, default=None):
+    """优先从 SQLite 读取，失败回退 JSON"""
+    try:
+        from mofin_db import get_conn
+        conn = get_conn()
+        result = query_func(conn)
+        conn.close()
+        if result:
+            return result
+    except Exception:
+        pass
+    return _load_json(json_path, default)
+
+
 def _save_json(path, data):
     os.makedirs(os.path.dirname(path), exist_ok=True)
     with open(path, "w", encoding="utf-8") as f:
@@ -49,46 +63,80 @@ def index():
 @app.route("/api/portfolio")
 def api_portfolio():
     """持仓列表"""
-    data = _load_json(DATA_DIR / "portfolio.json")
+    try:
+        from mofin_db import get_conn, query_holdings, query_portfolio_summary
+        conn = get_conn()
+        holdings = query_holdings(conn)
+        summary = query_portfolio_summary(conn)
+        conn.close()
+        if holdings:
+            data = dict(summary)
+            data["holdings"] = holdings
             return jsonify(data)
+    except Exception:
+        pass
+    return jsonify(_load_json(DATA_DIR / "portfolio.json"))
 
 
 @app.route("/api/watchlist")
 def api_watchlist():
     """自选列表"""
-    data = _load_json(DATA_DIR / "watchlist.json")
+    try:
-    return jsonify(data)
+        from mofin_db import get_conn, query_watchlist
+        conn = get_conn()
+        stocks = query_watchlist(conn)
+        conn.close()
+        if stocks:
+            return jsonify({"stocks": stocks})
+    except Exception:
+        pass
+    return jsonify(_load_json(DATA_DIR / "watchlist.json"))
 
 
 @app.route("/api/overview")
 def api_overview():
     """概览数据"""
+    try:
+        from mofin_db import get_conn, query_holdings, query_portfolio_summary, query_latest_market
+        conn = get_conn()
+        holdings = query_holdings(conn)
+        summary = query_portfolio_summary(conn)
+        market = query_latest_market(conn)
+        conn.close()
+        if holdings:
+            total_assets = summary.get("total_assets", 0) or 0
+            stock_value = summary.get("stock_value", 0) or 0
+            cash = summary.get("cash", 0) or 0
+            position_pct = summary.get("position_pct", 0) or 0
+            total_pnl = summary.get("total_pnl", 0) or 0
+            top_movers = sorted(
+                [h for h in holdings if abs(h.get("change_pct", 0) or 0) >= 3],
+                key=lambda x: abs(x.get("change_pct", 0) or 0), reverse=True)[:5]
+            return jsonify({
+                "total_assets": total_assets, "stock_value": stock_value,
+                "cash": cash, "position_pct": position_pct, "total_pnl": total_pnl,
+                "top_movers": top_movers, "market": market,
+                "alerts": _load_json(DATA_DIR / "alerts.json", [])[:10],
+                "updated_at": summary.get("updated_at", ""),
+            })
+    except Exception:
+        pass
     portfolio = _load_json(DATA_DIR / "portfolio.json", [])
     market = _load_json(DATA_DIR / "market.json", {})
     alerts = _load_json(DATA_DIR / "alerts.json", [])
-
     total_assets = portfolio.get("total_assets", 0)
     stock_value = portfolio.get("stock_value", 0)
     cash = portfolio.get("cash", 0)
     position_pct = portfolio.get("position_pct", 0)
     total_pnl = portfolio.get("total_pnl", 0)
     holdings = portfolio.get("holdings", [])
-
     top_movers = sorted(
         [h for h in holdings if abs(h.get("change_pct", 0)) >= 3],
-        key=lambda x: abs(x.get("change_pct", 0)),
+        key=lambda x: abs(x.get("change_pct", 0)), reverse=True)[:5]
-        reverse=True,
-    )[:5]
-
     return jsonify({
-        "total_assets": total_assets,
+        "total_assets": total_assets, "stock_value": stock_value,
-        "stock_value": stock_value,
+        "cash": cash, "position_pct": position_pct, "total_pnl": total_pnl,
-        "cash": cash,
+        "top_movers": top_movers, "market": market, "alerts": alerts[:10],
-        "position_pct": position_pct,
-        "total_pnl": total_pnl,
-        "top_movers": top_movers,
-        "market": market,
-        "alerts": alerts[:10],
         "updated_at": portfolio.get("updated_at", ""),
     })
 
@@ -138,8 +186,16 @@ def api_stock(code):
 @app.route("/api/market")
 def api_market():
     """市场观察"""
-    data = _load_json(DATA_DIR / "market.json", {})
+    try:
+        from mofin_db import get_conn, query_latest_market
+        conn = get_conn()
+        data = query_latest_market(conn)
+        conn.close()
+        if data and data.get("sectors"):
             return jsonify(data)
+    except Exception:
+        pass
+    return jsonify(_load_json(DATA_DIR / "market.json", {}))
 
 
 # ── 数据写入API（供 cron/update_data.py 调用） ──────────

strategy_feedback.py

@@ -175,6 +175,14 @@ def generate_adjustment(decision, phase_check, accuracy_trend):
 
 def run():
     decisions = load_json(DECISIONS_PATH, {"decisions": []})
+    # 优先从 SQLite 读取价格事件
+    try:
+        from mofin_db import get_conn, query_price_events
+        conn = get_conn()
+        pe_rows = query_price_events(conn, limit=50000)
+        conn.close()
+        events = {"events": pe_rows}
+    except Exception:
         events = load_json(EVENTS_PATH, {"events": []})
     accuracy_stats = load_json(ACCURACY_PATH, {})
 

strategy_lifecycle.py

@@ -77,11 +77,24 @@ def load_stock_sector_map():
     stock_sector_map.json 格式: {code: [sector1, sector2, ...]}
     跳过 _note, _created_at 等元数据键。
     """
+    # 优先从 SQLite 读取
+    try:
+        from mofin_db import get_conn, query_sector_stocks
+        conn = get_conn()
+        # 从 stock_sectors 表反向构建 code→[sectors] 映射
+        rows = conn.execute("SELECT code, sector_name FROM stock_sectors ORDER BY code").fetchall()
+        conn.close()
+        code_to_sectors = {}
+        for code, sector in rows:
+            if code not in code_to_sectors:
+                code_to_sectors[code] = []
+            code_to_sectors[code].append(sector)
+        return code_to_sectors
+    except Exception:
+        pass
     try:
         with open(STOCK_SECTOR_MAP_PATH) as f:
             data = json.load(f)
-        # stock_sector_map.json 直接是 code→[sectors] 格式
-        # 过滤掉 _note, _created_at 等元数据键
         code_to_sectors = {}
         for key, value in data.items():
             if key.startswith("_"):
@@ -94,13 +107,37 @@ def load_stock_sector_map():
 
 
 def load_market_context():
-    """读取市场上下文(market.json)，返回 (sector_perf, breadth, sentiment, mood)
+    """读取市场上下文，优先 SQLite，回退 market.json"""
-    
+    # 优先从 SQLite 读取
-    sector_perf: {sector_name: sector_data_dict} — 所有行业板块数据
+    try:
-    breadth: up_ratio 值（如27.8）
+        from mofin_db import get_conn, query_latest_market
-    sentiment: market.json 的 mood（bearish/neutral/bullish）
+        conn = get_conn()
-    top_gainers/losers: 涨幅/跌幅前5
+        market = query_latest_market(conn)
-    """
+        conn.close()
+        if market and market.get("sectors"):
+            sector_perf = {}
+            for s in market["sectors"]:
+                name = s.get("name", "")
+                if name:
+                    sector_perf[name] = {
+                        "change": s.get("change_pct", 0),
+                        "up_count": s.get("up_count", 0),
+                        "down_count": s.get("down_count", 0),
+                        "net_inflow": s.get("net_inflow", 0),
+                        "lead_stock": s.get("lead_stock", ""),
+                        "lead_stock_change": s.get("lead_stock_change", 0),
+                    }
+            return {
+                "sector_perf": sector_perf,
+                "breadth": market.get("up_ratio", 50),
+                "mood": market.get("mood", "neutral"),
+                "top_gainers": {g["name"]: g["change_pct"] for g in market.get("top_gainers", [])},
+                "top_losers": {g["name"]: g["change_pct"] for g in market.get("top_losers", [])},
+                "total_sectors": len(market["sectors"]),
+                "market_timestamp": market.get("timestamp", ""),
+            }
+    except Exception:
+        pass
     try:
         with open(MARKET_CONTEXT_PATH) as f:
             market = json.load(f)
@@ -117,7 +154,6 @@ def load_market_context():
                     "lead_stock": s.get("lead_stock", ""),
                     "lead_stock_change": s.get("lead_stock_change", 0),
                 }
-        # 涨幅/跌幅前5
         top_gainers = {s.get("name", ""): s.get("change", 0)
                        for s in market.get("top_gainers", [])}
         top_losers = {s.get("name", ""): s.get("change", 0)
@@ -896,9 +932,18 @@ def check_sector_alerts(market_ctx, stock_sector_map, holdings, wl):
 
 def regenerate_all(stdout=True):
     """全量重评所有持仓+自选策略"""
+    # 优先从 SQLite 读取
+    try:
+        from mofin_db import get_conn, query_holdings, query_watchlist
+        conn = get_conn()
+        holdings = query_holdings(conn)
+        wl_stocks = query_watchlist(conn)
+        conn.close()
+        pf = {"holdings": holdings}
+        wl = {"stocks": wl_stocks}
+    except Exception:
         pf = safe_json_load(PORTFOLIO_PATH, {})
-    wl_path = WATCHLIST_PATH
+        wl = safe_json_load(WATCHLIST_PATH, {})
-    wl = safe_json_load(wl_path, {})
 
     all_stocks = {}
     for item in pf.get("holdings", []):

system_health_check.py

@@ -110,13 +110,20 @@ def run():
     # 4. 价格事件统计
     lines.append("")
     lines.append("【价格事件】")
+    try:
+        from mofin_db import get_conn, query_price_events, query_price_events_by_date
+        conn = get_conn()
+        ev_list = query_price_events(conn, limit=50000)
+        today_events = query_price_events_by_date(conn, now.strftime("%Y-%m-%d"))
+        conn.close()
+    except Exception:
         events = load_json(EVENTS_PATH, {"events": []})
         ev_list = events.get("events", [])
         today_events = [e for e in ev_list if e.get("date") == now.strftime("%Y-%m-%d")]
     lines.append(check(len(ev_list) > 0, f"历史事件: {len(ev_list)}条"))
     lines.append(check(len(today_events) > 0, f"今日事件: {len(today_events)}条"))
     if len(ev_list) == 0:
-        issues.append("price_events.json 无事件记录，price_monitor可能未触发过")
+        issues.append("price_events 无事件记录，price_monitor可能未触发过")
         warn_count += 1
     else:
         ok_count += 1