fix(pipelines): clear today's real cron errors + kill monitoring false alarms

Real errors fixed (all verified by manual run):
- price_monitor.py: shares None -> TypeError at L584 (now completes 3m7s,
  full 39-stock reassess + zone triggers + Dad push)
- market_insight.py: net_inflow None -> TypeError at L142 (now 0.3s, 5 insights)
- promote_candidates.py: add busy_timeout=30s (DB lock under concurrent writes)
- premarket_full_review.py: 12-dim analysis now detached background launch
  (was doomed by cron 120s script timeout no matter what)

Systemic:
- HERMES_CRON_SCRIPT_TIMEOUT=600 drop-in for both gateway services
  (fixes mofin_health SIGTERM, market_watch timeout, memory_guardian timeout)
- sync_profile_scripts.sh: re-hardlink deploy->profile scripts after every
  deploy (scp replaces files = new inode = broken hardlink = cron silently
  runs stale code; this caused promote to keep failing after my first fix)

Monitoring false-alarm fixes (the '花瓶' problem):
- mofin_health.py: legacy JSONs that migrated to DB (multi_tf_cache/
  macro_context/market/live_prices/price_history/macro_risk_state) no longer
  warn 'no readers'; marked as migrated
- NEW db_freshness section: real pipeline health from DB tables
  (mtf_cache 0.4h / macro_context_log 2h / market_snapshots 2h /
  live_prices 0.4h / price_events.json 0.4h — ALL HEALTHY)
- price_events freshness reads live JSON store (DB table is legacy)
- market.json placeholder created (13+ scripts have fallback paths)

Investigation notes: wiki-self-growth 03:04 key1 429 predates full key6
activation on default gateway; current 8642 verified on key6 and working.
Weekend 'Blocked' jobs verified fixed (vacuum_state_db passes).

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

@@ -20,27 +20,22 @@ 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 深度分析（每日强制，14只约8-10分钟）
+# Step 1.5: 持仓 12 维 LLM 深度分析——后台分离执行（12-40分钟，不能阻塞 cron 的 120s 超时）
 print("\n" + "=" * 50)
-print("🧠 持仓12维LLM分析（每日强制刷新）")
+print("🧠 持仓12维LLM分析（后台分离启动）")
 print("=" * 50)
 import subprocess as _sp
-analysis_result = {"ok": 0, "fail": 0, "skip": 0}
+analysis_result = {"mode": "detached"}
 try:
-    r = _sp.run(
+    _log = open("/tmp/holdings_12d_daily.log", "a")
+    _sp.Popen(
         ["python3", "/home/hmo/.hermes/profiles/position-analyst/scripts/batch_reassess.py",
          "--type", "holding", "--today"],
-        capture_output=True, text=True, timeout=3600)
-    print(r.stdout[-2000:] if len(r.stdout) > 2000 else r.stdout)
-    if r.returncode != 0 and r.stderr:
-        print(f"⚠️ stderr: {r.stderr[:300]}")
-    # 从输出尾部解析统计
-    import re as _re
-    m = _re.search(r"完成: (\d+)成功, (\d+)失败, (\d+)跳过", r.stdout)
-    if m:
-        analysis_result = {"ok": int(m.group(1)), "fail": int(m.group(2)), "skip": int(m.group(3))}
+        stdout=_log, stderr=_log, start_new_session=True)
+    print("  ✅ 12维分析已后台启动，日志: /tmp/holdings_12d_daily.log（结果落DB，不阻塞盘前流程）")
 except Exception as e:
-    print(f"⚠️ 12维分析步骤异常: {e}")
+    print(f"  ⚠️ 12维分析启动失败: {e}")
+    analysis_result = {"mode": "detached", "error": str(e)[:100]}
 
 # Step 2: 自选退出
 print("\n" + "=" * 50)

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
     
     # 读未提拔候选（按评分降序）

deploy/profile-scripts/sync_profile_scripts.sh

@@ -0,0 +1,13 @@
+#!/bin/bash
+# sync_profile_scripts.sh — 把 deploy/profile-scripts 全部硬链接到 profile scripts 目录
+# 每次部署(scp/git)后必须跑：scp替换文件会破坏硬链接(inode变更)，导致cron跑旧版本。
+set -e
+SRC="/home/hmo/MoFin/deploy/profile-scripts"
+DST="/home/hmo/.hermes/profiles/position-analyst/scripts"
+count=0
+for f in "$SRC"/*.py; do
+    name=$(basename "$f")
+    ln -f "$f" "$DST/$name"
+    count=$((count+1))
+done
+echo "synced $count scripts (hardlink)"