Implement proposal #19: Score-and-Retry + JSON format validation for 知微 pipeline

- Create json_validator.py utility module:
  - validate_json_syntax(): JSON syntax scoring (0-100) with auto-fix
  - validate_with_schema(): Multi-layer validation (syntax + schema)
  - score_and_retry(): Generic retry wrapper with error feedback
  - safe_parse_structured(): Safe <structured_data> extraction
  - FormatErrorLibrary: Persistent error tracking + stats + auto recommendations
  - CLI entry point for standalone use

- Integrate into strategy_lifecycle.py:
  - safe_json_load() now uses json_validator for auto-fix + error logging
  - Falls back to legacy repair on ImportError

- Integrate into update_data.py:
  - parse_report() uses safe_parse_structured() instead of bare json.loads()
  - Errors logged to FormatErrorLibrary silently

- Convert strategy_lifecycle.py to Unix line endings
This commit is contained in:
知微
2026-07-08 01:18:48 +08:00
parent 073cc1a2de
commit 7cebcba13a
3 changed files with 3392 additions and 2589 deletions
+24 -12
View File
@@ -55,20 +55,32 @@ def parse_report(markdown_text, source_file=None):
report["type"] = "盘前"
# ★ 优先提取结构化JSON(如果知微输出了的话)
struct_match = re.search(r'<structured_data>\s*(\{.*?\})\s*</structured_data>', markdown_text, re.DOTALL)
struct_match = re.search(r'<structured_data>\s*(.*?)\s*</structured_data>', markdown_text, re.DOTALL)
if struct_match:
try:
parsed = json.loads(struct_match.group(1))
report["structured"] = parsed
# 从结构化数据中直接取stock codes
codes = set()
for h in parsed.get("holdings", []):
c = h.get("code", "")
if c:
codes.add(c)
report["stocks_mentioned"] = sorted(codes)
except (json.JSONDecodeError, Exception) as e:
pass # JSON解析失败→走NLP兜底
import sys as _sys
_sys.path.insert(0, "/home/hmo/MoFin")
from json_validator import safe_parse_structured
parsed_sd, sd_score, sd_errors = safe_parse_structured(
struct_match.group(0),
log_errors=True,
)
if parsed_sd:
report["structured"] = parsed_sd
report["_json_score"] = sd_score
# 从结构化数据中直接取stock codes
codes = set()
for h in parsed_sd.get("holdings", []):
c = h.get("code", "")
if c:
codes.add(c)
report["stocks_mentioned"] = sorted(codes)
elif sd_errors:
# 记录错误但继续,不阻断流程
pass
except Exception:
pass # 兜底,不阻断 pipeline
# 摘要(前3非空行)
body_lines = [l.strip() for l in lines if l.strip() and not l.strip().startswith("#") and not l.strip().startswith("##")]