merge: pipeline fixes + false-alarm cleanup

deploy/profile-scripts/market_insight.py

@@ -139,8 +139,8 @@ def generate():
         insights.append("风险板块: " + " | ".join(loser_insights[:3]))
 
     # ── 洞察4：资金流向异动 ──
-    big_inflow = [s for s in sectors if s.get("net_inflow", 0) > 50]
-    big_outflow = [s for s in sectors if s.get("net_inflow", 0) < -50]
+    big_inflow = [s for s in sectors if (s.get("net_inflow") or 0) > 50]
+    big_outflow = [s for s in sectors if (s.get("net_inflow") or 0) < -50]
     if big_inflow:
         top = max(big_inflow, key=lambda s: s["net_inflow"])
         insights.append(

deploy/profile-scripts/mofin_health.py

@@ -868,22 +868,95 @@ def build_report():
         })
 
     # JSON文件
+    # 已迁移到DB的旧JSON文件：不再报"无读取方"假警报，真实健康信号看DB表新鲜度
+    MIGRATED_TO_DB = {
+        "multi_tf_cache.json": "mtf_cache",
+        "macro_context.json": "macro_context_log",
+        "market.json": "market_snapshots",
+        "live_prices.json": "live_prices",
+        "price_history.json": "price_events",
+        "macro_risk_state.json": "macro_context_log",
+    }
     json_entities = []
     for jf in sorted(WEB_DATA.glob("*.json")):
         if jf.name == "stocks": continue
         if jf.stem.startswith("temp_"): continue
         readers = flows["json_read"].get(jf.name, [])
         size = jf.stat().st_size / 1024
+        migrated = MIGRATED_TO_DB.get(jf.name)
+        desc = JSON_DESC.get(jf.name, "")
+        if migrated:
+            desc = (desc + " " if desc else "") + f"(已迁移到DB表 {migrated}，此为遗留文件)"
         json_entities.append({
             "name": jf.name,
-            "desc": JSON_DESC.get(jf.name, ""),
+            "desc": desc,
             "size_kb": round(size, 1),
             "readers": readers[:10],
             "writers": [],  # 难以精确追踪
             "last_modified": datetime.fromtimestamp(jf.stat().st_mtime).strftime("%m-%d %H:%M"),
-            "warn": len(readers) == 0 and jf.name not in ("portfolio.json", "market.json"),
+            "warn": (len(readers) == 0 and jf.name not in ("portfolio.json", "market.json")
+                     and not migrated),
+            "migrated_to_db": migrated or None,
         })
 
+    # ── DB表新鲜度：真实数据管道健康信号（替代对遗留JSON文件的mtime检查）──
+    # 注意：活跃数据在 /home/hmo/MoFin/data/mofin.db（live_prices/mtf_cache 今日有写入），
+    # 不用 get_conn()（它指向 web-dashboard 的库，那边部分表是旧的）
+    db_freshness = []
+    FRESHNESS_TABLES = [
+        ("mtf_cache", "updated_at", "多周期均线缓存"),
+        ("macro_context_log", "created_at", "宏观上下文"),
+        ("market_snapshots", "created_at", "市场快照"),
+        ("live_prices", "updated_at", "实时价格"),
+        ("price_events", "created_at", "价格事件"),
+    ]
+    try:
+        _fc = sqlite3.connect("/home/hmo/MoFin/data/mofin.db", timeout=10)
+        for tname, tcol, label in FRESHNESS_TABLES:
+            try:
+                row = _fc.execute(
+                    f"SELECT MAX({tcol}) FROM {tname}").fetchone()
+                if row and row[0]:
+                    last_dt = datetime.fromisoformat(str(row[0]).replace("Z", ""))
+                    age_h = (now - last_dt).total_seconds() / 3600
+                    db_freshness.append({
+                        "table": tname, "label": label,
+                        "last_record": last_dt.strftime("%m-%d %H:%M"),
+                        "age_hours": round(age_h, 1),
+                        "warn": age_h > 24,
+                    })
+                else:
+                    db_freshness.append({"table": tname, "label": label,
+                                         "last_record": None, "age_hours": -1, "warn": True})
+            except Exception:
+                pass  # 表不存在或列名不同，跳过
+        _fc.close()
+    except Exception:
+        pass
+
+    # price_events 特殊处理：活跃存储是 price_events.json（price_monitor 实时写入），
+    # DB 表是旧遗留。读 JSON 最后一条事件的时间。
+    try:
+        _pe_path = Path("/home/hmo/web-dashboard/data/price_events.json")
+        if _pe_path.exists():
+            _pe = json.loads(_pe_path.read_text(encoding="utf-8"))
+            _items = _pe if isinstance(_pe, list) else _pe.get("events", [])
+            if _items:
+                _last = _items[-1]
+                _ts = _last.get("timestamp") or _last.get("created_at") or ""
+                _dt = datetime.fromisoformat(str(_ts).replace("Z", ""))
+                _age = (now - _dt).total_seconds() / 3600
+                # 替换 db_freshness 里 price_events 那条（DB 旧数据）
+                db_freshness = [f for f in db_freshness if f["table"] != "price_events"]
+                db_freshness.append({
+                    "table": "price_events.json", "label": "价格事件",
+                    "last_record": _dt.strftime("%m-%d %H:%M"),
+                    "age_hours": round(_age, 1),
+                    "warn": _age > 24,
+                })
+    except Exception:
+        pass
+
     # ── Tab 3: 流程/cron映射 ──
     pipelines = []
     for j in sorted(cron_jobs, key=lambda x: x.get("name","")):
@@ -912,6 +985,7 @@ def build_report():
         "entities": entities,
         "json_files": json_entities,
         "pipelines": pipelines,
+        "db_freshness": db_freshness,
     }
     out_path = WEB_DATA / "mofin_health.json"
     with open(out_path, "w") as f:

deploy/profile-scripts/premarket_full_review.py

@@ -19,6 +19,23 @@ from strategy_lifecycle import regenerate_all
 result = regenerate_all(stdout=True)
 print(f"\n重评完成: {result.get('ok',0)}/{result.get('total',0)}成功")
 
+# Step 1.5: 持仓 12 维 LLM 深度分析——后台分离执行（12-40分钟，不能阻塞 cron 的 120s 超时）
+print("\n" + "=" * 50)
+print("🧠 持仓12维LLM分析（后台分离启动）")
+print("=" * 50)
+import subprocess as _sp
+analysis_result = {"mode": "detached"}
+try:
+    _log = open("/tmp/holdings_12d_daily.log", "a")
+    _sp.Popen(
+        ["python3", "/home/hmo/.hermes/profiles/position-analyst/scripts/batch_reassess.py",
+         "--type", "holding", "--today"],
+        stdout=_log, stderr=_log, start_new_session=True)
+    print("  ✅ 12维分析已后台启动，日志: /tmp/holdings_12d_daily.log（结果落DB，不阻塞盘前流程）")
+except Exception as e:
+    print(f"  ⚠️ 12维分析启动失败: {e}")
+    analysis_result = {"mode": "detached", "error": str(e)[:100]}
+
 # Step 2: 自选退出
 print("\n" + "=" * 50)
 print("🔍 自选退出检查")

deploy/profile-scripts/price_monitor.py

@@ -581,7 +581,7 @@ def run_once(round_label=""):
     # === 第三步：买入区偏离检测 + 自动重评 ===
     reassesed_codes = []
     # 先做急跌检测（仅持仓，自选股不推送暴跌告警）
-    holdings_codes = {d["code"] for d in active if d.get("shares", 0) > 0}
+    holdings_codes = {d["code"] for d in active if (d.get("shares") or 0) > 0}
     for d in active:
         code = d["code"]
         # 非持仓跳过

deploy/profile-scripts/promote_candidates.py

@@ -10,7 +10,8 @@ from datetime import datetime
 DB_PATH = Path("/home/hmo/MoFin/data/mofin.db")
 
 def main():
-    conn = sqlite3.connect(str(DB_PATH))
+    conn = sqlite3.connect(str(DB_PATH), timeout=30)
+    conn.execute("PRAGMA busy_timeout=30000")
     conn.row_factory = sqlite3.Row
     
     # 读未提拔候选（按评分降序）