fix: DB路径大面积修复——mofin_db/server/technical_analysis/strategy_tree等脚本的Path(__file__).parent/data错误指向scripts/data而非data/ 导致读写分离

- 量价分析: full_analysis输出volume_deep+成交量存储在price_history.json
- FK约束移除: holding_strategies外键->holdings阻止自选股写入
- #000850 重评已写入(止损3.74/止盈4.06/RR1.67)含量价信号

scripts/clean_watchlist.py

@@ -9,8 +9,8 @@ sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 from mo_data import read_portfolio, read_decisions, read_watchlist
 from mofin_db import get_conn, write_watchlist_stock, write_holding_strategy
 
-WL = "/home/hmo/web-dashboard/data/watchlist.json"
-DEC = "/home/hmo/web-dashboard/data/decisions.json"
+WL = "/home/hmo/web-dashboard/data/watchlist.json"  # 路径保留用于历史备份兼容，数据实际走DB
+DEC = "/home/hmo/web-dashboard/data/decisions.json"  # 同上
 
 holding_codes = set()
 pf = read_portfolio()
@@ -33,8 +33,7 @@ removed = [s for s in stocks if s.get("code") in holding_codes]
 after = len(new_stocks)
 wl["stocks"] = new_stocks
 
-# Backup
-os.rename(WL, WL + ".bak2")
+# Backup — DB 版，不再碰JSON文件
 # DB 写入
 conn = get_conn()
 for s in wl.get("stocks", []):
@@ -61,8 +60,7 @@ for d in dec.get("decisions", []):
             dec_changed += 1
             
 if dec_changed:
-    os.rename(DEC, DEC + ".bak3")
-    # DB 写入
+    # DB 写入（不再碰JSON文件）
     conn = get_conn()
     for d in dec.get("decisions", []):
         write_holding_strategy(conn, d.get("code", ""), d.get("name", ""), d)

scripts/data/format_error_library.json

@@ -0,0 +1,35 @@
+{
+  "library_version": "1.0",
+  "total_records": 2,
+  "last_updated": "2026-07-08T01:14:29.611447",
+  "records": [
+    {
+      "task_id": "test-task",
+      "timestamp": "2026-07-08T01:14:29.582665",
+      "attempt": 1,
+      "score": 85,
+      "errors": [
+        "小问题"
+      ],
+      "is_failure": false,
+      "context": {
+        "code": "000001",
+        "name": "测试"
+      }
+    },
+    {
+      "task_id": "test-task",
+      "timestamp": "2026-07-08T01:14:29.611356",
+      "attempt": 2,
+      "score": 60,
+      "errors": [
+        "缺少字段: stop_loss",
+        "JSON语法错误"
+      ],
+      "is_failure": true,
+      "context": {
+        "code": "000001"
+      }
+    }
+  ]
+}

scripts/data/mofin.db

@@ -1 +0,0 @@
-/home/hmo/MoFin/data/mofin.db

scripts/fix_portfolio_prices.py

@@ -1,84 +1,78 @@
 #!/usr/bin/env python3
 """
-fix_portfolio_prices.py — 一次性修复脚本：从 decisions.json 读取 price 字段
-更新两个 canonical portfolio.json，并重算 total_assets/position_pct。
+fix_portfolio_prices.py — 一次性修复脚本：从 decisions(DB) 读取 price 字段
+更新 holdings 表，并重算 total_assets/position_pct。
 
 背景：strategy_lifecycle.regenerate_all() 在旧版本中会
 从 DB query_holdings()（不含 price/change_pct）覆盖写入
 portfolio.json，导致 price_monitor 维护的实时价丢失。
 该 bug 已在 strategy_lifecycle.py:1790 修复（保留 price 字段），
-此脚本用于修复已损坏的 portfolio.json。
+此脚本用于修复已损坏的 portfolio 数据。
 
 用法：
-  python3 fix_portfolio_prices.py          # 修复两个 canonical 文件
+  python3 fix_portfolio_prices.py          # 修复 DB
   python3 fix_portfolio_prices.py --check  # 只检查，不修改
 """
 
-import json
 import sys
-from pathlib import Path
 
-# Canonical paths
-MOFIN_PORTFOLIO = Path("/home/hmo/MoFin/data/portfolio.json")
-DASHBOARD_PORTFOLIO = Path("/home/hmo/web-dashboard/data/portfolio.json")
-DECISIONS_PATH = Path("/home/hmo/web-dashboard/data/decisions.json")
+from mo_data import read_decisions, read_portfolio
+from mofin_db import get_conn, write_holdings_batch, write_portfolio_summary
 
 
-def load_decisions():
-    """从 decisions.json 读取 {code: {price, ...}} 映射"""
+def build_price_map():
+    """从 decisions(DB) 读取 {code: {price, ...}} 映射"""
     try:
-        with open(DECISIONS_PATH) as f:
-            raw = json.load(f)
+        dec = read_decisions()
     except Exception as e:
-        print(f"❌ 无法读取 {DECISIONS_PATH}: {e}", file=sys.stderr)
+        print(f"❌ 无法读取 decisions(DB): {e}", file=sys.stderr)
         return {}
-    
+
     price_map = {}
-    for d in raw.get("decisions", []):
+    for d in dec.get("decisions", []):
         code = d.get("code", "")
         price = d.get("price", 0) or d.get("current_price", 0)
         if code and price:
             price_map[code] = float(price)
-    print(f"  decisions.json: {len(price_map)} 个股票的 price 已加载")
+    print(f"  decisions(DB): {len(price_map)} 个股票的 price 已加载")
     return price_map
 
 
-def fix_portfolio(path: Path, price_map: dict, check_only: bool):
-    """修复一个 portfolio.json 文件"""
-    if not path.exists():
-        print(f"❌ {path} 不存在，跳过")
+def fix_portfolio(check_only: bool) -> bool:
+    """从 DB 读持仓，修复 price，写回 DB"""
+    try:
+        pf = read_portfolio()
+    except Exception as e:
+        print(f"❌ 无法读取 portfolio(DB): {e}", file=sys.stderr)
         return False
 
-    try:
-        with open(path) as f:
-            pf = json.load(f)
-    except Exception as e:
-        print(f"❌ 无法读取 {path}: {e}", file=sys.stderr)
+    holdings = pf.get("holdings", [])
+    price_map = build_price_map()
+    if not price_map:
+        print("❌ decisions 无有效 price 数据，退出")
         return False
 
     changes = 0
     errors = 0
-    holdings = pf.get("holdings", [])
-    
+
     for h in holdings:
         code = h.get("code", "")
         if not code:
             continue
-        
+
         old_price = h.get("price", 0)
         decision_price = price_map.get(code)
-        
+
         if decision_price and (not old_price or old_price == 0):
             h["price"] = decision_price
             changes += 1
-            print(f"  ✅ {code} {h.get('name','')}: price {old_price} → {decision_price}")
+            print(f"  ✅ {code} {h.get('name','')}: price {old_price} \u2192 {decision_price}")
         elif decision_price and old_price and abs(decision_price - old_price) / max(abs(old_price), 1) > 0.05:
-            # Price differs >5% from decision - flag it but don't overwrite (price_monitor is authoritative)
             print(f"  ⚠️ {code} {h.get('name','')}: 当前价 {old_price} vs decisions {decision_price} (偏离>{5:.0f}%)，保留当前价")
         elif not decision_price:
             errors += 1
             if old_price == 0 or not old_price:
-                print(f"  ❌ {code} {h.get('name','')}: decisions.json 无此股 price 数据，当前 price={old_price}")
+                print(f"  ❌ {code} {h.get('name','')}: decisions 无此股 price 数据，当前 price={old_price}")
 
     # 重算 total_assets / position_pct
     total_mv = 0
@@ -86,38 +80,56 @@ def fix_portfolio(path: Path, price_map: dict, check_only: bool):
         price = h.get("price", 0) or 0
         shares = h.get("shares", 0) or 0
         total_mv += price * shares
-    
+
     cash = pf.get("cash", 0) or 0
-    new_total_assets = round(total_mv + cash, 2)
+    frozen = pf.get("frozen_cash", 0) or 0
+    new_total_assets = round(total_mv + cash + frozen, 2)
     new_position_pct = round(total_mv / new_total_assets * 100, 2) if new_total_assets > 0 else 0
-    
+
     old_total_assets = pf.get("total_assets", 0)
     old_position_pct = pf.get("position_pct", 0)
-    
+
     if abs(new_total_assets - old_total_assets) > 100:
-        print(f"  📊 total_assets: {old_total_assets} → {new_total_assets} (变动 {new_total_assets-old_total_assets:.0f})")
-        pf["total_assets"] = new_total_assets
+        print(f"  📊 total_assets: {old_total_assets} \u2192 {new_total_assets} (变动 {new_total_assets-old_total_assets:.0f})")
         changes += 1
-    
+
     if abs(new_position_pct - old_position_pct) > 0.5:
-        print(f"  📊 position_pct: {old_position_pct}% → {new_position_pct}%")
-        pf["position_pct"] = new_position_pct
+        print(f"  📊 position_pct: {old_position_pct}% \u2192 {new_position_pct}%")
         changes += 1
-    
-    pf["updated_at"] = __import__("datetime").datetime.now().strftime('%Y-%m-%d %H:%M')
 
     if changes == 0:
-        print(f"  ✅ {path.name}: 无需修复（{len(holdings)} 个持仓价格正常）")
+        print(f"  ✅ 无需修复（{len(holdings)} 个持仓价格正常）")
         return True
 
     if check_only:
         print(f"  ⏸️ 检查模式: {changes} 处需修复，未写入")
         return True
 
+    # 写入 DB
     try:
-        with open(path, "w") as f:
-            json.dump(pf, f, indent=2, ensure_ascii=False)
-        print(f"  ✅ {path.name}: {changes} 处已修复，已写入")
+        conn = get_conn()
+        ok, msg = write_holdings_batch(conn, holdings)
+        if not ok:
+            print(f"  ❌ 写入 holdings 失败: {msg}", file=sys.stderr)
+            conn.close()
+            return False
+
+        summary = {
+            "total_assets": new_total_assets,
+            "total_mv": round(total_mv, 2),
+            "stock_value": round(total_mv, 2),
+            "cash": cash,
+            "frozen_cash": frozen,
+            "position_pct": new_position_pct,
+            "currency": pf.get("currency", "CNY"),
+        }
+        ok, msg = write_portfolio_summary(conn, summary)
+        conn.close()
+        if not ok:
+            print(f"  ❌ 写入 portfolio_summary 失败: {msg}", file=sys.stderr)
+            return False
+
+        print(f"  ✅ DB: {changes} 处已修复，已写入")
         return True
     except Exception as e:
         print(f"  ❌ 写入失败: {e}", file=sys.stderr)
@@ -126,25 +138,16 @@ def fix_portfolio(path: Path, price_map: dict, check_only: bool):
 
 def main():
     check_only = "--check" in sys.argv
-    
+
     print("=== fix_portfolio_prices.py ===")
     if check_only:
         print("模式：检查（不写入）")
     else:
         print("模式：修复")
-    
-    price_map = load_decisions()
-    if not price_map:
-        print("❌ decisions.json 无有效 price 数据，退出")
-        return 1
-    
-    print(f"\n--- {MOFIN_PORTFOLIO.name} ---")
-    ok1 = fix_portfolio(MOFIN_PORTFOLIO, price_map, check_only)
-    
-    print(f"\n--- {DASHBOARD_PORTFOLIO.name} ---")
-    ok2 = fix_portfolio(DASHBOARD_PORTFOLIO, price_map, check_only)
-    
-    if ok1 and ok2:
+
+    ok = fix_portfolio(check_only)
+
+    if ok:
         print("\n✅ 全部完成")
         return 0
     else:

scripts/inspect_decisions.py

@@ -1,6 +1,5 @@
-import json
-with open('/home/hmo/web-dashboard/data/decisions.json') as f:
-    d = json.load(f)
+from mo_data import read_decisions
+d = read_decisions()
 for i, e in enumerate(d.get('decisions', [])[:3]):
     print(f"\n=== Entry {i}: {e.get('code')} {e.get('name')} ===")
     for k, v in sorted(e.items()):

scripts/intraday_health_check.py

@@ -241,7 +241,13 @@ def main():
     check_gateways()
     check_xiaoguo()
     if 9 <= now.hour < 16:
-        check_price_monitor()
+        # 开盘前10分钟（9:00-9:10）跳过价格新鲜度检查
+        # price_monitor 从 09:00 才开始启动，09:01 检查时数据还未更新（前一天收盘数据）
+        # 给 price_monitor 足够时间完成第一轮数据拉取更新
+        if now.hour == 9 and now.minute < 10:
+            log(True, "开盘初期，价格监控grace period（跳过新鲜度检查）")
+        else:
+            check_price_monitor()
     check_signal_pipeline()
 
     write_todos()

scripts/json_validation.py

@@ -0,0 +1,870 @@
+#!/usr/bin/env python3
+"""
+json_validation.py — JSON 格式校验 + 打分-重试循环 + Format Error Library
+
+为知微股票分析 Pipeline 提供三层校验：
+  1. JSON 语法解析 (json.loads)
+  2. JSON Schema 字段匹配（可选）
+  3. 语义完整性检查（关键字段非空、数值范围合理）
+
+核心功能：
+  - validate_json_syntax(raw) → (score, errors, parsed_or_None)
+  - scored_retry_loop(llm_fn, max_retries=3, threshold=90, schema=None)
+  - FormatErrorLibrary — 持久化错误记录，每日/周汇总
+  - best_effort_fallback(raw) → dict（尽力恢复）
+
+用法：
+  from json_validation import scored_retry_loop, FormatErrorLibrary
+  
+  def my_llm_call(prompt):
+      # 调用 LLM 返回原始文本
+      return llm_response
+  
+  result = scored_retry_loop(
+      llm_fn=my_llm_call,
+      prompt=initial_prompt,
+      context={"code": "000001", "name": "平安银行"},
+      schema=SCHEMA,
+      max_retries=3,
+      threshold=90,
+  )
+  # result = {"ok": True, "data": {...}, "attempts": 2, "score": 95}  或
+  # result = {"ok": False, "data": {...}, "attempts": 3, "score": 60, "fallback": True}
+"""
+
+import json
+import os
+import re
+import sys
+import time
+from datetime import datetime, timedelta
+from pathlib import Path
+from collections import defaultdict
+
+# ─── 常量 ───────────────────────────────────────────────────
+DEFAULT_THRESHOLD = 90       # 验收阈值（0-100）
+DEFAULT_MAX_RETRIES = 3      # 最大重试次数
+ERROR_LIBRARY_PATH = os.path.join(
+    os.path.dirname(os.path.abspath(__file__)),
+    "data", "format_error_library.json"
+)
+
+# ─── 标准 JSON Schema 种子字段 ─────────────────────────────
+# 知微股票分析输出期望包含的字段
+STOCK_ANALYSIS_SCHEMA = {
+    "required": [
+        "code", "name", "action", "stop_loss", "take_profit",
+        "entry_low", "entry_high", "timing_signal", "rr_ratio",
+    ],
+    "optional": [
+        "price", "cost", "shares", "currency",
+        "tech_snapshot", "signal_factors", "sector_context",
+        "multi_tf_context", "status", "quality_check",
+    ],
+    "types": {
+        "code": str, "name": str, "action": str,
+        "stop_loss": (int, float), "take_profit": (int, float),
+        "entry_low": (int, float), "entry_high": (int, float),
+        "timing_signal": str, "rr_ratio": (int, float),
+    }
+}
+
+
+# ═══════════════════════════════════════════════════════════
+# 第一层：JSON 语法校验 + 打分
+# ═══════════════════════════════════════════════════════════
+
+def validate_json_syntax(raw_text):
+    """JSON 语法校验 + 打分（0-100）
+    
+    返回: (score, errors, parsed_or_None)
+      score: 0-100 整数
+      errors: 错误描述列表
+      parsed_or_None: 成功 parse 的 dict 或 None
+    """
+    errors = []
+    
+    if not raw_text or not raw_text.strip():
+        return (0, ["空文本"], None)
+    
+    text = raw_text.strip()
+    
+    # 步骤1: 尝试提取可能的 JSON（从 markdown 代码块中）
+    extracted = _extract_json_block(text)
+    if extracted != text:
+        text = extracted
+    
+    # 步骤2: 初步语法检查
+    score = 50  # 基础分
+    
+    # 检查是否有 JSON 的基本特征
+    if text.startswith("{") or text.startswith("["):
+        score += 10
+    else:
+        errors.append("不以 { 或 [ 开头，可能不是 JSON")
+    
+    if text.endswith("}") or text.endswith("]"):
+        score += 10
+    else:
+        errors.append("不以 } 或 ] 结尾，可能被截断")
+    
+    # 步骤3: 尝试解析
+    try:
+        parsed = json.loads(text)
+        score += 30  # 成功解析加 30 分
+        
+        # 检查深层嵌套完整性
+        _check_nested_integrity(parsed, errors)
+        
+        # 最终得分
+        if not errors:
+            score = min(100, score + 10)  # 完美通过再加 10
+        
+        return (min(100, score), errors, parsed)
+    
+    except json.JSONDecodeError as e:
+        errors.append(f"JSON 语法错误: {e}")
+        
+        # 常见错误类型归类
+        err_msg = str(e)
+        if "trailing comma" in err_msg.lower() or "extra data" in err_msg.lower():
+            errors.append("类型: 尾随逗号或多余数据")
+            score -= 5
+        elif "unterminated" in err_msg.lower():
+            errors.append("类型: 未闭合的引号/括号")
+            score -= 10
+        elif "unexpected" in err_msg.lower():
+            errors.append("类型: 意外的字符/标记")
+            score -= 8
+        elif "control character" in err_msg.lower():
+            errors.append("类型: 字符串包含未转义的控制字符")
+            score -= 5
+        elif "expect value" in err_msg.lower():
+            errors.append("类型: 缺少值或字段名")
+            score -= 8
+        
+        # 尝试自动修复
+        fixed, fixed_desc = _auto_fix_json(text)
+        if fixed:
+            try:
+                parsed = json.loads(fixed)
+                score = max(score, 75)  # 修复成功给 75 分
+                errors.append(f"自动修复成功: {fixed_desc}")
+                return (score, errors, parsed)
+            except json.JSONDecodeError:
+                pass
+        
+        return (max(0, score - 20), errors, None)
+
+
+def _extract_json_block(text):
+    """从文本中提取 JSON 代码块（```json ... ```）"""
+    # 模式1: ```json ... ``` 或 ``` ... ```
+    m = re.search(r'```(?:json)?\s*(\{.*?\}|\[.*?\])\s*```', text, re.DOTALL)
+    if m:
+        return m.group(1).strip()
+    # 模式2: <structured_data> ... </structured_data>
+    m = re.search(r'<structured_data>\s*(\{.*?\}|\[.*?\])\s*</structured_data>', text, re.DOTALL)
+    if m:
+        return m.group(1).strip()
+    return text
+
+
+def _check_nested_integrity(obj, errors, path=""):
+    """递归检查嵌套结构的完整性"""
+    if isinstance(obj, dict):
+        for k, v in obj.items():
+            child_path = f"{path}.{k}" if path else k
+            if v is None:
+                errors.append(f"字段 {child_path} 为 null")
+            elif isinstance(v, str) and len(v) > 1000:
+                errors.append(f"字段 {child_path} 字符串过长 ({len(v)} 字符)")
+            else:
+                _check_nested_integrity(v, errors, child_path)
+    elif isinstance(obj, list):
+        if len(obj) > 100:
+            errors.append(f"数组 {path} 过长 ({len(obj)} 项)")
+        for i, item in enumerate(obj):
+            _check_nested_integrity(item, errors, f"{path}[{i}]")
+
+
+def _auto_fix_json(text):
+    """尝试自动修复常见 JSON 格式问题
+    
+    返回: (fixed_text_or_None, description_or_None)
+    """
+    original = text
+    fixes = []
+    
+    # 修复1: 去掉多余尾部括号（可能有多余的 }}})
+    fixed = text.rstrip()
+    # 计算开括号和闭括号的数量
+    open_braces = fixed.count('{')
+    close_braces = fixed.count('}')
+    open_brackets = fixed.count('[')
+    close_brackets = fixed.count(']')
+    
+    if close_braces > open_braces:
+        # 去掉多余的 }
+        excess = close_braces - open_braces
+        for _ in range(excess):
+            last_idx = fixed.rfind('}')
+            if last_idx >= 0:
+                fixed = fixed[:last_idx] + fixed[last_idx+1:]
+        fixes.append(f"去掉 {excess} 个多余右括号")
+    
+    if close_brackets > open_brackets:
+        excess = close_brackets - open_brackets
+        for _ in range(excess):
+            last_idx = fixed.rfind(']')
+            if last_idx >= 0:
+                fixed = fixed[:last_idx] + fixed[last_idx+1:]
+        fixes.append(f"去掉 {excess} 个多余右中括号")
+    
+    # 修复2: 字符串内未转义的换行符
+    result = []
+    in_str = False
+    for ch in fixed:
+        if ch == '"':
+            in_str = not in_str
+            result.append(ch)
+        elif in_str and ch in '\n\r':
+            result.append('\\n')
+        else:
+            result.append(ch)
+    fixed2 = ''.join(result)
+    if fixed2 != fixed:
+        fixes.append("转义字符串内换行符")
+        fixed = fixed2
+    
+    # 修复3: 尾随逗号（在 } 或 ] 前的逗号）
+    fixed = re.sub(r',\s*}', '}', fixed)
+    fixed = re.sub(r',\s*\]', ']', fixed)
+    if fixed != original:
+        fixes.append("移除尾随逗号")
+    
+    # 修复4: 如果还不行，尝试补全为有效的对象包装
+    if not fixed.startswith('{') and not fixed.startswith('['):
+        # 可能是裸值 → 包装为对象
+        try:
+            json.loads(fixed)
+        except json.JSONDecodeError:
+            pass  # 保留原始
+    
+    if fixed != original:
+        return fixed, '; '.join(fixes)
+    
+    # 无有效修复
+    return None, None
+
+
+# ═══════════════════════════════════════════════════════════
+# 第二层：JSON Schema 校验
+# ═══════════════════════════════════════════════════════════
+
+def validate_json_schema(parsed, schema=None):
+    """JSON Schema 字段匹配验证
+    
+    schema 格式:
+      {
+          "required": ["field1", "field2"],
+          "optional": ["field3"],
+          "types": {"field1": str, "field2": (int, float)},
+          "min_values": {"rr_ratio": 0},
+          "non_empty": ["action", "timing_signal"],
+      }
+    
+    返回: (score, errors)
+      score: 0-100
+      errors: 字段级别的错误列表
+    """
+    if schema is None:
+        schema = STOCK_ANALYSIS_SCHEMA
+    
+    if not isinstance(parsed, dict):
+        return (30, ["顶层不是对象（dict），无法做 Schema 校验"])
+    
+    errors = []
+    score = 50  # 基础分
+    
+    required = schema.get("required", [])
+    optional = schema.get("optional", [])
+    types = schema.get("types", {})
+    min_values = schema.get("min_values", {})
+    non_empty = schema.get("non_empty", [])
+    ranges = schema.get("ranges", {})  # {"field": (min, max)}
+    
+    # 检查必要字段
+    missing_required = []
+    for field in required:
+        if field not in parsed:
+            missing_required.append(field)
+    
+    if missing_required:
+        errors.append(f"缺少必要字段: {', '.join(missing_required)}")
+        score -= len(missing_required) * 8
+    else:
+        score += 15
+    
+    # 检查字段类型
+    type_errors = []
+    for field, expected_type in types.items():
+        if field in parsed and parsed[field] is not None:
+            val = parsed[field]
+            if not isinstance(val, expected_type):
+                type_errors.append(f"{field}: 期望 {expected_type.__name__}, 实际 {type(val).__name__}")
+    
+    if type_errors:
+        errors.extend(type_errors)
+        score -= len(type_errors) * 5
+    
+    # 检查最小值约束
+    for field, min_val in min_values.items():
+        if field in parsed and parsed[field] is not None:
+            try:
+                if float(parsed[field]) < min_val:
+                    errors.append(f"{field}={parsed[field]} < 最小值 {min_val}")
+                    score -= 5
+            except (TypeError, ValueError):
+                pass
+    
+    # 检查非空约束
+    for field in non_empty:
+        if field in parsed:
+            val = parsed[field]
+            if val is None or (isinstance(val, str) and not val.strip()) or val == 0:
+                errors.append(f"{field} 为空/零")
+                score -= 5
+    
+    # 检查数值范围
+    for field, (lo, hi) in ranges.items():
+        if field in parsed and parsed[field] is not None:
+            try:
+                v = float(parsed[field])
+                if v < lo or v > hi:
+                    errors.append(f"{field}={v} 超出合理范围 [{lo}, {hi}]")
+                    score -= 3
+            except (TypeError, ValueError):
+                pass
+    
+    # 未知字段不扣分（可接受额外信息）
+    
+    # 标准化 score
+    score = max(0, min(100, score))
+    return (score, errors)
+
+
+# ═══════════════════════════════════════════════════════════
+# 第三层：打分-重试循环
+# ═══════════════════════════════════════════════════════════
+
+def scored_retry_loop(llm_fn, prompt, context=None, schema=None,
+                      max_retries=3, threshold=90,
+                      error_lib=None, task_id=None):
+    """打分-重试循环
+    
+    参数:
+      llm_fn: callable(prompt) → raw_text
+              每次重试时传入附加上一轮反馈的增强 prompt
+      prompt: 初始 prompt 文本
+      context: dict，额外的上下文信息，用于记录
+      schema: JSON Schema 定义（可选）
+      max_retries: 最大重试次数（默认 3）
+      threshold: 验收阈值 0-100（默认 90）
+      error_lib: FormatErrorLibrary 实例（可选），用于记录错误
+      task_id: 任务标识符（可选），用于错误库跟踪
+    
+    返回:
+      {
+          "ok": True/False,        # True=在阈值内通过, False=最终fallback
+          "data": dict or None,    # 解析后的 JSON 数据
+          "raw": str,              # 最终原始输出
+          "attempts": int,         # 实际尝试次数
+          "scores": [int, ...],    # 每次尝试的分数
+          "errors": [str, ...],    # 最终错误列表  
+          "fallback": bool,        # 是否走了一段 best_effort
+      }
+    """
+    # 验证最小报价字段
+    _validate_min_values(schema)
+    
+    best_score = 0
+    best_data = None
+    best_raw = ""
+    
+    # 构建基础 prompt 系统反馈部分
+    base_prompt = prompt
+    feedback_history = []
+    scores = []
+    
+    for attempt in range(1, max_retries + 1):
+        # 构建本次的 prompt（附加上一轮的格式反馈）
+        current_prompt = base_prompt
+        if feedback_history:
+            feedback_section = "\n\n[Format Feedback from previous attempt]\n" + \
+                               "\n".join(feedback_history)
+            current_prompt = base_prompt + feedback_section
+        
+        # 调用 LLM
+        try:
+            raw = llm_fn(current_prompt)
+        except Exception as e:
+            scores.append(0)
+            feedback_history.append(f"Attempt {attempt}: LLM call failed: {e}")
+            continue
+        
+        if not raw or not raw.strip():
+            scores.append(0)
+            feedback_history.append(f"Attempt {attempt}: 空响应")
+            continue
+        
+        # 步骤1: 语法校验
+        syntax_score, syntax_errors, parsed = validate_json_syntax(raw)
+        
+        # 步骤2: Schema 校验（如果语法通过且有 schema）
+        schema_score = 100
+        schema_errors = []
+        if parsed is not None and schema:
+            schema_score, schema_errors = validate_json_schema(parsed, schema)
+        
+        # 综合打分: 语法占 60%, Schema 占 40%
+        if parsed is not None:
+            combined_score = int(syntax_score * 0.6 + schema_score * 0.4)
+        else:
+            combined_score = syntax_score  # 语法失败时只用语法分
+        
+        scores.append(combined_score)
+        
+        # 记录错误信息
+        all_errors = syntax_errors + schema_errors
+        
+        # 记录到 Format Error Library
+        if error_lib and task_id:
+            error_lib.record(
+                task_id=task_id,
+                attempt=attempt,
+                score=combined_score,
+                errors=all_errors,
+                context=context,
+            )
+        
+        # 追踪最佳结果
+        if combined_score > best_score:
+            best_score = combined_score
+            best_data = parsed
+            best_raw = raw
+        
+        # 检查是否通过阈值
+        if combined_score >= threshold:
+            return {
+                "ok": True,
+                "data": parsed,
+                "raw": raw,
+                "attempts": attempt,
+                "scores": scores,
+                "errors": all_errors,
+                "fallback": False,
+            }
+        
+        # 准备下一轮反馈
+        if attempt < max_retries:
+            feedback = _build_feedback(combined_score, all_errors, attempt)
+            feedback_history.append(feedback)
+        
+        # 小幅延迟避免 API 限流
+        if attempt < max_retries:
+            time.sleep(0.5)
+    
+    # 所有重试都失败 → 尝试 best_effort_fallback
+    fallback_data = best_data
+    fallback_raw = best_raw
+    
+    if fallback_data is None and fallback_raw:
+        # 尝试用 best_effort 从最佳原始输出恢复
+        fallback_data = best_effort_fallback(fallback_raw)
+    
+    # 构建最终结果
+    result = {
+        "ok": fallback_data is not None,
+        "data": fallback_data,
+        "raw": fallback_raw,
+        "attempts": max_retries,
+        "scores": scores,
+        "errors": ["所有重试均未达到阈值", f"最高分: {best_score}/{threshold}"],
+        "fallback": True,
+    }
+    
+    # 产生告警（通过 stderr 输出，可被 cron 捕获）
+    task_tag = f"[{task_id}]" if task_id else ""
+    print(
+        f"[JSON_VALIDATION] WARNING{task_tag}: "
+        f"{max_retries}次重试后最高分{best_score}/{threshold}，"
+        f"使用{'fallback数据' if fallback_data else '空数据'}",
+        file=sys.stderr, flush=True
+    )
+    
+    # 记录严重错误到 error_lib
+    if error_lib and task_id:
+        error_lib.record(
+            task_id=task_id,
+            attempt=max_retries + 1,
+            score=best_score,
+            errors=["ALL_RETRIES_EXHAUSTED", f"最高分{best_score}/{threshold}"],
+            context=context,
+            is_failure=True,
+        )
+    
+    return result
+
+
+def _validate_min_values(schema):
+    """确保 schema 包含 min_values 约束"""
+    if schema and "min_values" not in schema:
+        schema["min_values"] = {}
+    return schema
+
+
+def _build_feedback(score, errors, attempt):
+    """构建格式反馈字符串（给 LLM 的下一轮提示）"""
+    parts = [f"--- Attempt {attempt} feedback (score: {score}/100) ---"]
+    if errors:
+        parts.append("Issues to fix:")
+        for e in errors[:5]:  # 最多反馈 5 个问题
+            parts.append(f"  - {e}")
+    else:
+        parts.append("No specific errors found, but score below threshold.")
+    
+    return "\n".join(parts)
+
+
+# ═══════════════════════════════════════════════════════════
+# Best-effort Fallback
+# ═══════════════════════════════════════════════════════════
+
+def best_effort_fallback(raw_text):
+    """尽力从非法 JSON 中恢复数据
+    
+    策略（按优先级）:
+      1. 尝试 auto_fix 后解析
+      2. 正则提取所有 key:value 对
+      3. 提取股票代码等关键信息
+    """
+    if not raw_text:
+        return None
+    
+    # 策略1: auto_fix
+    fixed, _ = _auto_fix_json(raw_text)
+    if fixed:
+        try:
+            return json.loads(fixed)
+        except json.JSONDecodeError:
+            pass
+    
+    # 策略2: 正则提取所有字段
+    data = {}
+    
+    # 提取代码
+    code_match = re.search(r'"code"\s*:\s*"(\d+)"', raw_text)
+    if code_match:
+        data["code"] = code_match.group(1)
+    
+    # 提取名称
+    name_match = re.search(r'"name"\s*:\s*"([^"]+)"', raw_text)
+    if name_match:
+        data["name"] = name_match.group(1)
+    
+    # 提取数值字段
+    for field in ["stop_loss", "take_profit", "entry_low", "entry_high", 
+                   "price", "rr_ratio", "cost"]:
+        pattern = rf'"{field}"\s*:\s*([\d.]+)'
+        m = re.search(pattern, raw_text)
+        if m:
+            try:
+                data[field] = float(m.group(1))
+            except ValueError:
+                pass
+    
+    # 提取字符串字段
+    for field in ["action", "timing_signal", "currency", "status"]:
+        pattern = rf'"{field}"\s*:\s*"([^"]*)"'
+        m = re.search(pattern, raw_text)
+        if m:
+            data[field] = m.group(1)
+    
+    return data if data else None
+
+
+# ═══════════════════════════════════════════════════════════
+# Format Error Library
+# ═══════════════════════════════════════════════════════════
+
+class FormatErrorLibrary:
+    """格式错误库 — 持久化记录每次格式错误
+    
+    每月/周自动汇总，支持按错误类型和任务ID查询。
+    """
+    
+    def __init__(self, path=None):
+        self.path = path or ERROR_LIBRARY_PATH
+        self._records = []
+        self._load()
+    
+    def _load(self):
+        """从磁盘加载已有记录"""
+        if os.path.exists(self.path):
+            try:
+                with open(self.path, "r", encoding="utf-8") as f:
+                    data = json.load(f)
+                    self._records = data.get("records", [])
+            except (json.JSONDecodeError, Exception):
+                self._records = []
+        else:
+            self._records = []
+    
+    def _save(self):
+        """持久化到磁盘"""
+        os.makedirs(os.path.dirname(self.path), exist_ok=True)
+        with open(self.path, "w", encoding="utf-8") as f:
+            json.dump({
+                "library_version": "1.0",
+                "total_records": len(self._records),
+                "last_updated": datetime.now().isoformat(),
+                "records": self._records[-1000:],  # 最多保留最近1000条
+            }, f, ensure_ascii=False, indent=2)
+    
+    def record(self, task_id, attempt, score, errors, 
+               context=None, is_failure=False):
+        """记录一条格式错误"""
+        record = {
+            "task_id": task_id,
+            "timestamp": datetime.now().isoformat(),
+            "attempt": attempt,
+            "score": score,
+            "errors": errors[:10],  # 最多保留10个错误
+            "is_failure": is_failure,
+        }
+        if context:
+            # 只记录上下文的关键信息，避免敏感数据
+            safe_ctx = {}
+            for k in ["code", "name", "stock"]:
+                if k in (context or {}):
+                    safe_ctx[k] = context[k]
+            if safe_ctx:
+                record["context"] = safe_ctx
+        
+        self._records.append(record)
+        self._save()
+    
+    def get_summary(self, days=7):
+        """获取最近 N 天的错误汇总
+        
+        返回: {
+            "total": 总记录数,
+            "by_type": {"missing_field": N, "syntax_error": N, ...},
+            "by_task": {"task_id": N},
+            "worst_tasks": [最差任务列表],
+        }
+        """
+        cutoff = datetime.now() - timedelta(days=days)
+        recent = [
+            r for r in self._records 
+            if datetime.fromisoformat(r["timestamp"]) >= cutoff
+        ]
+        
+        if not recent:
+            return {"total": 0, "by_type": {}, "by_task": {}, "worst_tasks": []}
+        
+        by_type = defaultdict(int)
+        by_task = defaultdict(int)
+        task_scores = defaultdict(list)
+        
+        for r in recent:
+            task = r.get("task_id", "unknown")
+            by_task[task] += 1
+            task_scores[task].append(r.get("score", 0))
+            
+            for err in r.get("errors", []):
+                # 归类错误类型
+                if "缺少" in err or "missing" in err.lower():
+                    by_type["missing_field"] += 1
+                elif "syntax" in err.lower() or "JSON" in err or "json" in err:
+                    by_type["syntax_error"] += 1
+                elif "trailing comma" in err.lower() or "尾随逗号" in err:
+                    by_type["trailing_comma"] += 1
+                elif "truncat" in err.lower() or "截断" in err:
+                    by_type["truncation"] += 1
+                else:
+                    by_type["other"] += 1
+        
+        # 最差任务（平均分最低）
+        worst = sorted(
+            [(t, sum(s)/len(s), by_task[t]) for t, s in task_scores.items()],
+            key=lambda x: x[1]
+        )[:5]
+        
+        return {
+            "total": len(recent),
+            "period_days": days,
+            "by_type": dict(by_type),
+            "by_task": dict(by_task),
+            "worst_tasks": [
+                {"task": t, "avg_score": round(avg, 1), "count": c}
+                for t, avg, c in worst
+            ],
+            "generated_at": datetime.now().isoformat(),
+        }
+    
+    def get_recent(self, limit=20):
+        """获取最近的错误记录"""
+        return list(reversed(self._records[-limit:]))
+    
+    def clear_old(self, keep_days=30):
+        """清理超过 keep_days 的旧记录"""
+        cutoff = datetime.now() - timedelta(days=keep_days)
+        before = len(self._records)
+        self._records = [
+            r for r in self._records
+            if datetime.fromisoformat(r["timestamp"]) >= cutoff
+        ]
+        after = len(self._records)
+        self._save()
+        return before - after
+    
+    def auto_remediate_prompt(self):
+        """自动分析最近错误模式，返回改进提示词的建议
+        
+        返回: dict {建议类型: 建议内容}
+        """
+        summary = self.get_summary(days=7)
+        suggestions = {}
+        
+        by_type = summary.get("by_type", {})
+        total = summary.get("total", 0)
+        
+        if total == 0:
+            return {"info": "最近7天无格式错误，系统运行良好"}
+        
+        # 字段缺失问题 → 建议在 prompt 中显式列出字段
+        missing = by_type.get("missing_field", 0)
+        if missing > 0 and missing / max(total, 1) > 0.3:
+            suggestions["prompt_field_emphasis"] = (
+                f"最近7天 {missing}/{total} 错误是字段缺失，"
+                "建议在 prompt 中显式列出所有必需字段的说明和顺序"
+            )
+        
+        # 语法错误 → 建议使用结构化生成模式
+        syntax = by_type.get("syntax_error", 0)
+        if syntax > 0 and syntax / max(total, 1) > 0.5:
+            suggestions["structured_generation"] = (
+                f"最近7天 {syntax}/{total} 错误是 JSON 语法错误，"
+                "建议启用 constrained decoding 或 JSON mode"
+            )
+        
+        # 截断 → 检查 max_tokens
+        trunc = by_type.get("truncation", 0)
+        if trunc > 0:
+            suggestions["increase_max_tokens"] = (
+                f"最近7天 {trunc} 次截断，建议 max_tokens 从当前值上调"
+            )
+        
+        if not suggestions:
+            suggestions["info"] = "错误率在可接受范围内，无需自动调整"
+        
+        return suggestions
+
+
+# ═══════════════════════════════════════════════════════════
+# 便捷封装: 在 LLM 响应中提取 JSON 的统一入口
+# ═══════════════════════════════════════════════════════════
+
+def extract_json_safe(llm_response, schema=None, task_id=None):
+    """安全地从 LLM 响应中提取 JSON，带重试
+    
+    这是一个单轮快速提取版本（不做重试），适用于不需要
+    完整 scored_retry_loop 的场景。
+    
+    返回: (parsed_dict_or_None, score, errors)
+    """
+    score, errors, parsed = validate_json_syntax(llm_response)
+    
+    if parsed is not None and schema:
+        schema_score, schema_errors = validate_json_schema(parsed, schema)
+        score = int(score * 0.6 + schema_score * 0.4)
+        errors.extend(schema_errors)
+    
+    if parsed is None:
+        # 尝试 fallback
+        parsed = best_effort_fallback(llm_response)
+    
+    return parsed, score, errors
+
+
+# ═══════════════════════════════════════════════════════════
+# CLI 入口（测试/调试用）
+# ═══════════════════════════════════════════════════════════
+
+if __name__ == "__main__":
+    import sys
+    
+    if len(sys.argv) > 1 and sys.argv[1] == "--summary":
+        lib = FormatErrorLibrary()
+        summary = lib.get_summary(days=7)
+        print(json.dumps(summary, ensure_ascii=False, indent=2))
+        sys.exit(0)
+    
+    if len(sys.argv) > 1 and sys.argv[1] == "--recent":
+        lib = FormatErrorLibrary()
+        recent = lib.get_recent(limit=10)
+        print(json.dumps(recent, ensure_ascii=False, indent=2))
+        sys.exit(0)
+    
+    if len(sys.argv) > 1 and sys.argv[1] == "--test":
+        # 自测：验证各种情况
+        test_cases = [
+            ('{"code": "000001", "name": "平安"}', "标准 JSON"),
+            ('{"code": "000001", "name": "平安",}', "尾随逗号"),
+            ('```json\n{"code": "000001"}\n```', "代码块包裹"),
+            ('<structured_data>{"code": "000001"}</structured_data>', "structured_data 标签"),
+            ('{"code": "000001",\n"name": "平\n安"}', "未转义换行符"),
+            ('{"code": "000001"}}', "多余右括号"),
+            ('{"code": "000001", "price": 12.5, "stop_loss": null, "action": ""}', "空字段"),
+        ]
+        
+        for text, desc in test_cases:
+            score, errors, parsed = validate_json_syntax(text)
+            status = "✅" if parsed else "❌"
+            print(f"{status} {desc}: score={score}")
+            if errors:
+                for e in errors:
+                    print(f"   - {e}")
+            if parsed:
+                print(f"   parsed: {json.dumps(parsed, ensure_ascii=False)[:80]}")
+            print()
+        
+        # 测试 Schema 校验
+        print("=== Schema 校验测试 ===")
+        test_obj = {
+            "code": "000001", "name": "平安", "action": "持有",
+            "stop_loss": 10.5, "take_profit": 12.0,
+            "entry_low": 10.0, "entry_high": 11.0,
+            "timing_signal": "持有", "rr_ratio": 2.0,
+        }
+        score, errors = validate_json_schema(test_obj)
+        print(f"完整对象: score={score}, errors={errors}")
+        
+        test_obj2 = {"code": "000001", "name": "平安"}
+        score2, errors2 = validate_json_schema(test_obj2)
+        print(f"缺失字段: score={score2}, errors={errors2}")
+        
+        sys.exit(0)
+    
+    # 默认：从 stdin 读取 JSON 文本并验证
+    text = sys.stdin.read()
+    score, errors, parsed = validate_json_syntax(text)
+    result = {
+        "score": score,
+        "errors": errors,
+        "parsed": parsed,
+        "valid": parsed is not None,
+    }
+    print(json.dumps(result, ensure_ascii=False, indent=2))

scripts/market_insight.py

@@ -25,8 +25,9 @@ def load_holding_industry_map():
         with open(DATA_DIR / "stock_profiles.json", "r", encoding="utf-8") as f:
             profiles = json.load(f).get("profiles", [])
 
-        with open(DATA_DIR / "portfolio.json", "r", encoding="utf-8") as f:
-            portfolio = json.load(f)
+        # 优先从DB读取持仓（Dad铁律：禁用JSON直读）
+        from mo_data import read_portfolio
+        portfolio = read_portfolio()
     except FileNotFoundError:
         return {}
 
@@ -55,6 +56,8 @@ def load_holding_industry_map():
 
 
 def generate():
+    # market_path 在DB和fallback两个分支后都会用到，所以提前定义
+    market_path = DATA_DIR / "market.json"
     # 优先从 SQLite 读取市场数据
     try:
         from mofin_db import get_conn, query_latest_market

scripts/mo_alphasift_bridge.py

@@ -18,8 +18,8 @@ from pathlib import Path
 
 DSA_API = "http://127.0.0.1:8001"
 MOFIN_DATA = Path("/home/hmo/web-dashboard/data")
-WATCHLIST_PATH = MOFIN_DATA / "watchlist.json"
-PORTFOLIO_PATH = MOFIN_DATA / "portfolio.json"
+# 已迁移到 DB — watchlist.json / portfolio.json 不再使用
+# 保留路径仅用于兼容 import，实际数据从 mofin.db 读取
 
 DEFAULT_STRATEGIES = "balanced_alpha,dual_low,quality_value"
 DEFAULT_MARKET = "cn"
@@ -54,14 +54,18 @@ def api(endpoint, method="GET", body=None):
         return None
 
 def get_existing_codes():
+    """从 DB 读取已有持仓+自选。JSON 已废弃。"""
     codes = set()
-    for path in [WATCHLIST_PATH, PORTFOLIO_PATH]:
-        data = load_json(path)
-        if not data: continue
-        key = "stocks" if "watchlist" in str(path) else "holdings"
-        for item in data.get(key, []):
-            c = str(item.get("code", "")).strip()
-            if c: codes.add(c)
+    try:
+        import sqlite3
+        db = sqlite3.connect(str(MOFIN_DATA / "mofin.db"))
+        for row in db.execute("SELECT code FROM watchlist_stocks WHERE is_active=1"):
+            codes.add(str(row[0]).strip())
+        for row in db.execute("SELECT code FROM holdings"):
+            codes.add(str(row[0]).strip())
+        db.close()
+    except Exception as e:
+        print(f"WARN: DB读取失败: {e}")
     return codes
 
 
@@ -205,12 +209,19 @@ def run_all(strategies_str, market, max_results, dry_run=False):
         print("\n[DRY RUN] 未写入")
         return
     
-    # 写入
-    wl = load_json(WATCHLIST_PATH) or {"stocks": []}
-    wl["stocks"].extend(new_stocks)
-    wl["updated_at"] = time_str
-    save_json(WATCHLIST_PATH, wl)
-    print(f"\n已写入 {WATCHLIST_PATH}")
+    # 写入 DB
+    try:
+        sys.path.insert(0, str(MOFIN_DATA.parent))
+        from mofin_db import get_conn, write_watchlist_stock
+        conn = get_conn()
+        for s in new_stocks:
+            s.setdefault('currency', 'CNY')
+            write_watchlist_stock(conn, s)
+        conn.close()
+        print(f"\n已写入 {len(new_stocks)} 只到 DB watchlist_stocks")
+    except Exception as e:
+        print(f"WARN: DB写入失败: {e}")
+        return
     
     # 策略生成
     print("\n调用 regenerate_all()...")

scripts/mo_config.py

@@ -42,22 +42,26 @@ class MoConfig:
     # Hermes 状态目录
     hermes_dir: Path = field(default_factory=lambda: Path.home() / ".hermes")
     
-    # ── 关键数据文件路径 ──────────────────────────────────────────
-    
+    # ── 关键数据文件路径（已废弃，仅保留为检查逻辑。新代码勿用） ──────
+
     @property
     def portfolio_path(self) -> Path:
+        """⚠️ 已废弃！数据在 mofin.db holdings + portfolio_summary 表。"""
         return self.data_dir / "portfolio.json"
-    
+
     @property
     def decisions_path(self) -> Path:
+        """⚠️ 已废弃！数据在 mofin.db holding_strategies 表。"""
         return self.data_dir / "decisions.json"
-    
+
     @property
     def watchlist_path(self) -> Path:
+        """⚠️ 已废弃！数据在 mofin.db watchlist_stocks 表。"""
         return self.data_dir / "watchlist.json"
-    
+
     @property
     def price_events_path(self) -> Path:
+        """⚠️ 已废弃！数据在 mofin.db price_events 表。"""
         return self.data_dir / "price_events.json"
     
     @property

scripts/mofin_collect.py

@@ -155,11 +155,28 @@ try:
         for wr in conn.execute("SELECT code, name, price, entry_low, entry_high, stop_loss FROM watchlist_stocks WHERE is_active=1"):
             code = wr["code"]
             name = wr["name"]
+            wl_price = wr["price"] or 0
+            # 自选股price可能为0（新加入未更新），从实时API获取
+            if wl_price <= 0:
+                try:
+                    import urllib.request
+                    mkt = "hk" if len(str(code)) == 5 else "sh" if str(code)[0] in "56" else "sz"
+                    url = f"http://qt.gtimg.cn/q={mkt}{code}"
+                    req = urllib.request.Request(url, headers={'User-Agent': 'Mozilla/5.0'})
+                    resp = urllib.request.urlopen(req, timeout=5).read()
+                    text = resp.decode("gbk")
+                    parts = text.split("~")
+                    if len(parts) > 3:
+                        p = float(parts[3])
+                        if p > 0:
+                            wl_price = p
+                except Exception:
+                    pass
             # 自选股无cost/shares，传0
             try:
                 from strategy_lifecycle import reassess_with_context
                 result = reassess_with_context(
-                    code, name, wr["price"] or 0,
+                    code, name, wl_price,
                     0, 0, ""
                 )
                 if result and result.get("action"):

scripts/mofin_db.py

@@ -21,7 +21,7 @@ from datetime import datetime
 from pathlib import Path
 from typing import Optional, Callable
 
-DATA_DIR = Path(__file__).parent / "data"
+DATA_DIR = Path(__file__).parent.parent / "data"
 DB_PATH = DATA_DIR / "mofin.db"
 
 # ═══════════════════════════════════════════════════════════
@@ -1097,7 +1097,7 @@ def write_holding_strategy(conn, code: str, name: str, data: dict) -> tuple[bool
              signal_factors_json, time_horizon, decision_type)
             VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,
                     datetime('now','localtime'),
-                    ?,?,?,?,?,?,?,?,?)
+                    ?,?,?,?,?,?,?,?,?,?)
         """, (
             code, name,
             data.get('version', 1), data.get('price'), data.get('cost'),
@@ -1187,7 +1187,7 @@ def write_portfolio_summary(conn, data: dict) -> tuple[bool, str]:
 
 
 def write_watchlist_stock(conn, stock: dict) -> tuple[bool, str]:
-    """写入自选股（替代 watchlist.json）"""
+    """写入自选股（写入 watchlist_stocks 表）"""
     try:
         conn.execute("""
             INSERT INTO watchlist_stocks (code, name, price, entry_low, entry_high,

scripts/morning_health_check.py

@@ -672,7 +672,7 @@ def run_check(item):
                             if 'macro_risk_state' in content:
                                 consumer_info = 'macro_risk_state.json'
                             if 'watchlist' in content.lower():
-                                consumer_info = 'watchlist.json / decisions.json'
+                                consumer_info = 'watchlist_stocks表 / holding_strategies表'
                             if 'INSERT INTO' in content:
                                 for tbl in ['todos', 'price_events', 'macro_context_log', 'accuracy_stats']:
                                     if tbl in content:

scripts/multi_timeframe.py

@@ -17,7 +17,7 @@ import urllib.error
 from datetime import datetime, date, timedelta
 from typing import Optional
 
-DATA_DIR = "/home/hmo/web-dashboard/data"
+DATA_DIR = "/home/hmo/MoFin/data"
 HISTORY_PATH = os.path.join(DATA_DIR, "price_history.json")
 # multi_tf_cache.json 已迁移到 DB (mtf_cache 表)
 
@@ -92,7 +92,7 @@ def _load_mtf_cache():
         return _MTF_CACHE_DATA
     try:
         import sqlite3
-        db = sqlite3.connect('/home/hmo/web-dashboard/data/mofin.db')
+        db = sqlite3.connect('/home/hmo/MoFin/data/mofin.db')
         rows = db.execute("SELECT code, cache_json FROM mtf_cache").fetchall()
         _MTF_CACHE_DATA = {}
         for code, json_str in rows:
@@ -113,7 +113,7 @@ def _save_mtf_cache():
         return
     try:
         import sqlite3
-        db = sqlite3.connect('/home/hmo/web-dashboard/data/mofin.db')
+        db = sqlite3.connect('/home/hmo/MoFin/data/mofin.db')
         for code, data in _MTF_CACHE_DATA.items():
             db.execute(
                 "INSERT OR REPLACE INTO mtf_cache (code, cache_json, updated_at) VALUES (?,?,datetime('now','localtime'))",
@@ -222,9 +222,21 @@ def calc_moving_averages(klines: list, windows: list = [5, 10, 20, 60]) -> dict:
         return {f"ma{w}": None for w in windows}
 
     # 确保按时间正序（旧的在前）
+    # 使用日期判断顺序（不能用价格：下跌趋势下closes[0]>closes[-1]也会触发反转）
     closes = [k["close"] for k in klines]
-    # 检查是否倒序（最新的在前）
-    if len(closes) >= 2 and closes[0] > closes[-1]:
+    is_reversed = False
+    if len(klines) >= 2:
+        d0 = klines[0].get("date", "")
+        d1 = klines[-1].get("date", "")
+        if d0 and d1:
+            from datetime import datetime
+            try:
+                is_reversed = datetime.strptime(d0, "%Y-%m-%d") > datetime.strptime(d1, "%Y-%m-%d")
+            except:
+                # 日期格式不对时的fallback: 仅当首价显著高于末价才判定倒序(50%阈值)
+                is_reversed = (closes[0] > closes[-1] * 1.5) if len(closes) >= 2 else False
+
+    if is_reversed:
         closes = list(reversed(closes))
 
     result = {}

scripts/per_stock_reassess.py

@@ -172,16 +172,10 @@ def main():
             print(f"[ERROR] {code}: {e}", file=sys.stderr)
             errors += 1
 
-    # 写回 decisions.json（只更新被修改的那条，其余保留原样）
-    raw["decisions"] = list(decisions_map.values())
-    raw["total"] = len(raw["decisions"])
-    from datetime import datetime
-    raw["regenerated_at"] = datetime.now().strftime("%Y-%m-%d %H:%M")
-    with open(DECISIONS_PATH, "w") as f:
-        json.dump(raw, f, ensure_ascii=False, indent=2)
-
+    # 策略数据已通过DB写入（holding_strategies表），json.dump到decisions.json已废弃
     # 同步自选股更新回 watchlist_stocks 表
     try:
+        from datetime import datetime as _dt
         import sqlite3
         _db2 = sqlite3.connect('/home/hmo/web-dashboard/data/mofin.db')
         for _code in codes:
@@ -203,7 +197,7 @@ def main():
                         "stop_loss": _entry.get("stop_loss", 0),
                         "tech_snapshot": _entry.get("tech_snapshot", ""),
                         "rr": _entry.get("rr_ratio", 0),
-                        "reassessed_at": datetime.now().strftime("%Y-%m-%d")
+                        "reassessed_at": _dt.now().strftime("%Y-%m-%d")
                     }, ensure_ascii=False),
                     _code
                 ))

scripts/pre-flight-check.py

@@ -25,10 +25,10 @@ import sys
 from pathlib import Path
 from datetime import datetime, timezone
 
+from mo_data import read_decisions
+
 # === 路径 ===
 REGISTRY = Path("/home/hmo/projects/MoFin/data/prompts/registry.json")
-DECISIONS = Path("/home/hmo/web-dashboard/data/decisions.json")
-PORTFOLIO = Path("/home/hmo/web-dashboard/data/portfolio.json")
 
 # === 检查 7：版本一致性 ===
 def check_version_consistency():
@@ -62,33 +62,39 @@ def check_version_consistency():
 
 # === 检查 1：数据时效 ===
 def check_data_freshness():
-    """检查 decisions.json 的更新时间是否在合理范围内"""
+    """检查 holding_strategies 表的最新更新时间"""
+    from datetime import datetime, timezone
     try:
-        mtime = DECISIONS.stat().st_mtime
-        mtime_dt = datetime.fromtimestamp(mtime, tz=timezone.utc)
-        now = datetime.now(timezone.utc)
-        age_minutes = (now - mtime_dt).total_seconds() / 60
+        import sqlite3
+        c = sqlite3.connect(str(WEB_DATA / "mofin.db"))
+        row = c.execute("SELECT MAX(updated_at) FROM holding_strategies WHERE status IN ('active','updated')").fetchone()
+        c.close()
+        if row and row[0]:
+            mtime_dt = datetime.strptime(row[0][:19], '%Y-%m-%d %H:%M:%S')
+            age_minutes = (datetime.now() - mtime_dt).total_seconds() / 60
+        else:
+            return ("⚠️ 数据时效", "策略表无有效数据")
     except Exception as e:
-        return ("⚠️ 数据时效", f"无法检查 decisions.json: {e}")
+        return ("⚠️ 数据时效", f"无法检查策略表: {e}")
 
     if age_minutes < 60:
         return ("✅ 数据时效",
-                f"decisions.json 更新于 {age_minutes:.0f} 分钟前 ({mtime_dt.strftime('%H:%M')})")
+                f"策略数据 更新于 {age_minutes:.0f} 分钟前 ({mtime_dt.strftime('%H:%M')})")
     elif age_minutes < 240:
         return ("⚠️ 数据时效",
-                f"decisions.json 已 {age_minutes:.0f} 分钟未更新（可能已收盘），数据视为陈旧")
+                f"策略数据 已 {age_minutes:.0f} 分钟未更新（可能已收盘），数据视为陈旧")
     else:
         return ("❌ 数据时效",
-                f"decisions.json 已 {age_minutes:.0f} 分钟未更新，数据过期，请检查数据管线")
+                f"策略数据 已 {age_minutes:.0f} 分钟未更新，数据过期，请检查数据管线")
 
 
 # === 检查 5：成本有效 ===
 def check_cost_validity():
     """检查 decisions.json 中所有持仓的成本是否有效"""
     try:
-        dec = json.loads(DECISIONS.read_text())
+        dec = read_decisions()
     except Exception as e:
-        return ("⚠️ 成本有效性", f"无法读取 decisions.json: {e}")
+        return ("⚠️ 成本有效性", f"无法读取 decisions: {e}")
 
     stocks = dec.get("stocks", dec.get("holdings", dec.get("strategies", [])))
     if not stocks:
@@ -122,9 +128,9 @@ def check_cost_validity():
 def check_stop_technical(code):
     """检查单只股票的止损是否基于技术位（仅对单股模式生效）"""
     try:
-        dec = json.loads(DECISIONS.read_text())
+        dec = read_decisions()
     except Exception as e:
-        return ("⚠️ 止损技术位", f"无法读取 decisions.json: {e}")
+        return ("⚠️ 止损技术位", f"无法读取 decisions: {e}")
 
     stocks = dec.get("stocks", dec.get("strategies", []))
     for s in stocks:
@@ -150,9 +156,9 @@ def check_stop_technical(code):
 def check_rr(code, price=None):
     """检查单只股票的 R/R 是否达标"""
     try:
-        dec = json.loads(DECISIONS.read_text())
+        dec = read_decisions()
     except Exception as e:
-        return ("⚠️ R/R 达标", f"无法读取 decisions.json: {e}")
+        return ("⚠️ R/R 达标", f"无法读取 decisions: {e}")
 
     stocks = dec.get("stocks", dec.get("strategies", []))
     for s in stocks:

scripts/price_data_inject.py

@@ -6,6 +6,8 @@ import json
 import sys
 import os
 
+from mo_data import read_portfolio, read_watchlist
+
 PORTFOLIO_PATH = "/home/hmo/web-dashboard/data/portfolio.json"
 WATCHLIST_PATH = "/home/hmo/web-dashboard/data/watchlist.json"
 MTF_CACHE_PATH = "/home/hmo/web-dashboard/data/multi_tf_cache.json"
@@ -160,7 +162,7 @@ def classify_from_cache(code, name, mtf_cache):
 
 def main():
     try:
-        pf = json.load(open(PORTFOLIO_PATH))
+        pf = read_portfolio()
     except Exception as e:
         print(f"[ERROR] {e}")
         sys.exit(0)
@@ -220,7 +222,7 @@ def main():
 
     # ===== 自选股买入区监控 =====
     try:
-        wl = json.load(open(WATCHLIST_PATH))
+        wl = read_watchlist()
         wl_stocks = wl.get("stocks", [])
         # 过滤掉已经在持仓里的
         held_codes = {h["code"] for h in holdings}

scripts/price_monitor.py

@@ -3,7 +3,6 @@
 规则：进入区间报一次，离开区间报一次，中间不重复。
 每次运行时一次性刷新所有持仓+自选股的实时价。
 """
-import json
 import urllib.request
 import os
 import sys
@@ -11,9 +10,8 @@ import time
 import sqlite3
 from datetime import datetime
 
-DECISIONS_PATH = "/home/hmo/web-dashboard/data/decisions.json"
-PORTFOLIO_PATH = "/home/hmo/web-dashboard/data/portfolio.json"
-WATCHLIST_PATH = "/home/hmo/web-dashboard/data/watchlist.json"
+from mo_data import read_decisions
+
 BREACH_PATH = "/home/hmo/.hermes/zone_breach.json"
 STATE_PATH = os.path.expanduser("~/.hermes/price_trigger_state.json")
 EVENTS_PATH = "/home/hmo/web-dashboard/data/price_events.json"
@@ -319,10 +317,9 @@ def run_once(round_label=""):
 
     # === 第二步：检查触发条件 ===
     try:
-        with open(DECISIONS_PATH) as f:
-            dec = json.load(f)
+        dec = read_decisions()
     except:
-        print(f"❌{label} 无法读取decisions.json", file=sys.stderr)
+        print(f"❌{label} 无法读取decisions(DB)", file=sys.stderr)
         return
 
     active = [d for d in dec.get("decisions", []) if d.get("status") == "active"]

scripts/server.py

@@ -21,7 +21,7 @@ from mofin_db import get_conn, write_holdings_batch, write_portfolio_summary, wr
 
 app = Flask(__name__, static_folder="static", static_url_path="")
 
-DATA_DIR = Path(__file__).parent / "data"
+DATA_DIR = Path(__file__).parent.parent / "data"
 UPLOAD_DIR = Path(__file__).parent / "uploads"
 
 # Hermes Gateway
@@ -1038,7 +1038,7 @@ def update_realtime():
             pf_holdings[code]["updated_at"] = datetime.now().isoformat()
             updated += 1
 
-    # 也更新 watchlist.json
+    # 也更新 watchlist_stocks 表(DB)
     wl = read_watchlist()
     wl_stocks = {s["code"]: s for s in wl.get("stocks", [])}
 

scripts/strategy_summary.py

@@ -1,11 +1,9 @@
 #!/usr/bin/env python3
 """生成策略评估摘要"""
-import json
+from mo_data import read_decisions, read_portfolio
 
-with open('/home/hmo/web-dashboard/data/decisions.json') as f:
-    dec = json.load(f)
-with open('/home/hmo/MoFin/data/portfolio.json') as f:
-    pf = json.load(f)
+dec = read_decisions()
+pf = read_portfolio()
 
 holdings = pf.get('holdings', [])
 cash = pf.get('cash', 321271)

scripts/strategy_tree.py

@@ -74,7 +74,7 @@ def detect_scenario():
         # 优先 DB
         import sqlite3
         from pathlib import Path
-        db = sqlite3.connect(str(Path(__file__).parent / "data" / "mofin.db"))
+        db = sqlite3.connect(str(Path(__file__).parent.parent / "data" / "mofin.db"))
         mrow = db.execute(
             "SELECT indices, structure, sector_mood FROM macro_context_log "
             "WHERE has_valid_data=1 ORDER BY created_at DESC LIMIT 1"

scripts/system_audit.py

@@ -202,33 +202,18 @@ def audit_pipeline():
         except Exception as e:
             log_issue("数据管道", "HIGH", f"{name} 检查失败: {e}")
 
-    # 特殊检查：holding_strategies和decisions.json是否为空
+    # 检查 holding_strategies 表策略数量
     try:
-        hs = conn.execute("SELECT COUNT(*) FROM holding_strategies").fetchone()[0]
-        if hs == 0:
-            log_issue("数据管道", "HIGH", "holding_strategies表为空(策略评估产出未写入)", fix="检查holding_strategies写入逻辑")
+        hs_count = conn.execute("SELECT COUNT(*) FROM holding_strategies WHERE status IN ('active','updated')").fetchone()[0]
+        if hs_count < 5:
+            log_issue("数据管道", "HIGH", f"holding_strategies 仅{hs_count}条策略(异常)", fix="检查策略写入逻辑")
         else:
-            log_ok("数据管道", f"holding_strategies {hs}条")
+            log_ok("数据管道", f"holding_strategies {hs_count}条策略")
     except Exception as e:
         log_issue("数据管道", "HIGH", f"holding_strategies检查失败: {e}")
 
     conn.close()
-
-    # 检查decisions.json文件
-    try:
-        import json
-        with open(WEB_DATA / "decisions.json") as f:
-            dec = json.load(f)
-        cnt = len(dec.get("decisions", []))
-        if cnt < 5:
-            log_issue("数据管道", "HIGH", f"decisions.json仅{cnt}条决策(异常)", fix="检查decisions.json写入逻辑")
-        else:
-            log_ok("数据管道", f"decisions.json {cnt}条决策")
-    except Exception as e:
-        log_issue("数据管道", "HIGH", f"decisions.json读取失败: {e}")
-
-
-# ── 7. 系统服务 ──
+    
 def audit_services():
     services = [
         ("Dashboard", "http://127.0.0.1:8899/", "200"),

scripts/system_health_check.py

@@ -90,15 +90,15 @@ def run():
     try:
         from mo_data import read_portfolio, read_decisions, read_watchlist
         pf = read_portfolio()
-        lines.append(check(len(pf.get("holdings", [])) > 0, f"portfolio.json DB记录: {len(pf.get('holdings', []))}条"))
+        lines.append(check(len(pf.get("holdings", [])) > 0, f"持仓 DB记录: {len(pf.get('holdings', []))}条"))
         ok_count += 1
         wl = read_watchlist()
-        lines.append(check(len(wl.get("stocks", [])) > 0, f"watchlist.json DB记录: {len(wl.get('stocks', []))}条"))
+        lines.append(check(len(wl.get("stocks", [])) > 0, f"自选股 DB记录: {len(wl.get('stocks', []))}条"))
         ok_count += 1
         dec = read_decisions()
         # 注意：holding_strategies 表是决策的DB版本
         dec_count = len(dec.get("decisions", []))
-        lines.append(check(dec_count > 0, f"decisions.json 记录: {dec_count}条（DB holding_strategies表: {len(pf.get('holdings',[]))}只）"))
+        lines.append(check(dec_count > 0, f"DB策略记录: {dec_count}条"))
         ok_count += 1
     except Exception:
         lines.append(check(False, "MoFin DB 数据读取失败"))

scripts/technical_analysis.py

@@ -472,7 +472,7 @@ def analyze_volume_deep(code):
     import sqlite3
     from pathlib import Path
 
-    DATA_DIR = Path(__file__).parent / "data"
+    DATA_DIR = Path(__file__).parent.parent / "data"
     try:
         conn = sqlite3.connect(str(DATA_DIR / "mofin.db"))
         row = conn.execute("SELECT cache_json FROM mtf_cache WHERE code=?", (code,)).fetchone()

scripts/xiaoguo_signal_consumer.py

@@ -246,7 +246,7 @@ def main():
         if action == "watchlist":
             # 加自选
             results.append(f"✅ {sector_name}({code}): {summary}")
-            # 写入 watchlist.json
+            # 写入 watchlist_stocks 表(DB)
             try:
                 wl = read_watchlist()
                 wl.setdefault("stocks", [])