MoFin/json_validator.py

#!/usr/bin/env python3
"""
json_validator.py — 知微股票分析 Pipeline JSON 格式校验器

为「打分-重试 + 格式校验」双重循环提供基础设施。

核心功能：
1. validate_json_syntax(text) → (score, errors, parsed)
   json.loads() + 详细错误分类（语法错误/缺失字段/截断）

2. validate_with_schema(text, schema_fn) → (score, errors, parsed)
   在语法校验上叠Schema校验：字段存在性、类型正确性、非空约束

3. score_and_retry(llm_generate_fn, max_retries=3, threshold=90) → result
   通用重试包装器：不合格自动重试 + 错误反馈注入

4. FormatErrorLibrary — 格式错误持久化档案库
   记录每次错误 → 按类型统计 → 触发自动改进建议
"""

import json
import os
import re
import sys
import time
from datetime import datetime, timedelta
from pathlib import Path
from collections import defaultdict, Counter


# ── 错误类型分类 ──────────────────────────────────────────────

ERROR_SYNTAX = "syntax_error"       # JSON 语法解析失败
ERROR_MISSING_FIELD = "missing_field"  # Schema 必填字段缺失
ERROR_WRONG_TYPE = "wrong_type"      # 字段类型不符
ERROR_EMPTY_FIELD = "empty_field"    # 关键字段为空/null
ERROR_TRUNCATION = "truncation"      # 输出截断（在 JSON 外检测）
ERROR_UNEXPECTED = "unexpected_error"  # 其他


# ── 错误档案库路径 ────────────────────────────────────────────

FORMAT_ERROR_LIBRARY_PATH = Path(
    os.environ.get(
        "FORMAT_ERROR_LIBRARY_PATH",
        "/home/hmo/MoFin/data/format_error_library.json"
    )
)


# ═══════════════════════════════════════════════════════════════
# 核心校验器
# ═══════════════════════════════════════════════════════════════

def validate_json_syntax(text):
    """JSON 语法校验 + 评分。

    Args:
        text: 要校验的字符串（可能包含 ```json ... ``` 或纯 JSON）

    Returns:
        (score, errors, parsed)
        score: 0-100，100=完美JSON，<90=需要重试
        errors: [{'type': str, 'detail': str, 'line': int}, ...]
        parsed: 解析成功时返回 dict/list，失败返回 None
    """
    errors = []
    parsed = None

    if not text or not text.strip():
        errors.append({
            "type": ERROR_SYNTAX,
            "detail": "空输出",
            "line": 0,
        })
        return 0, errors, None

    # 尝试提取 JSON 块（支持 ```json ... ``` 包裹）
    raw = text.strip()
    json_block = raw

    # 检测 markdown 代码块
    code_match = re.search(
        r'```(?:json)?\s*\n?(.*?)```', raw, re.DOTALL
    )
    if code_match:
        json_block = code_match.group(1).strip()

    # 检测 <structured_data>...</structured_data> 包裹
    struct_match = re.search(
        r'<structured_data>\s*(.*?)\s*</structured_data>', raw, re.DOTALL
    )
    if struct_match:
        json_block = struct_match.group(1).strip()

    # 尝试括号定位：从第一个 { 到最后一个 }
    brace_start = json_block.find("{")
    brace_end = json_block.rfind("}")
    if brace_start >= 0 and brace_end > brace_start:
        json_block = json_block[brace_start:brace_end + 1]

    # 预处理修复常见问题
    fixed = _auto_fix_json(json_block)

    # 尝试解析
    try:
        parsed = json.loads(fixed)
    except json.JSONDecodeError as e:
        line_no = e.lineno if hasattr(e, 'lineno') else 0
        col_no = e.colno if hasattr(e, 'colno') else 0
        msg = str(e)

        # 分类错误类型
        err_type = _classify_syntax_error(msg, json_block)

        errors.append({
            "type": err_type,
            "detail": f"L{line_no}:{col_no} {msg[:200]}",
            "line": line_no,
        })

        # 尝试 best-effort 修复后再次解析
        repaired = _deep_fix_json(json_block)
        if repaired:
            try:
                parsed = json.loads(repaired)
                errors.append({
                    "type": "repaired",
                    "detail": f"自动修复后解析成功 (原始: {msg[:100]})",
                    "line": 0,
                })
            except json.JSONDecodeError:
                pass

    # 评分
    score = _compute_score(errors, parsed, text)

    return score, errors, parsed


def _auto_fix_json(text):
    """浅层自动修复常见 JSON 问题。"""
    if not text or not text.strip():
        return text

    fixed = text

    # 修复1: 尾随逗号在数组/对象末尾
    fixed = re.sub(r',\s*}', '}', fixed)
    fixed = re.sub(r',\s*\]', ']', fixed)

    # 修复2: 单引号代替双引号（只在键名位置）
    # 匹配 'key': 或 'key' :
    fixed = re.sub(r"'([^']+)'(\s*:)", r'"\1"\2', fixed)

    # 修复3: 字符串内未转义的换行符
    result = []
    in_str = False
    escape = False
    for ch in fixed:
        if escape:
            result.append(ch)
            escape = False
            continue
        if ch == '\\':
            result.append(ch)
            escape = True
            continue
        if ch == '"' and not escape:
            in_str = not in_str
            result.append(ch)
            continue
        if in_str and ch in '\n\r':
            result.append('\\n')
        else:
            result.append(ch)

    fixed = ''.join(result)

    # 修复4: 多余尾部括号
    fixed = fixed.rstrip('}') + '}'

    return fixed


def _deep_fix_json(text):
    """深度修复：尝试补全截断的 JSON。"""
    if not text:
        return None

    fixed = text.strip()

    # 如果是截断的（以不完整的状态结尾）
    # 尝试补全括号
    open_braces = fixed.count('{') - fixed.count('}')
    open_brackets = fixed.count('[') - fixed.count(']')

    if open_braces > 0:
        fixed += '}' * open_braces
    if open_brackets > 0:
        fixed += ']' * open_brackets

    # 如果缺少闭合引号
    in_str = False
    for ch in fixed:
        if ch == '"':
            in_str = not in_str
    if in_str:
        fixed += '"'

    # 尝试补全末尾逗号导致的问题
    fixed = re.sub(r',$', '', fixed.rstrip())

    # 补齐尾部的 }]
    if fixed.rstrip().endswith(','):
        fixed = fixed.rstrip()[:-1]

    # 确保对象闭合
    if '{' in fixed and '}' not in fixed[fixed.rfind('{'):]:
        fixed += '}'

    try:
        json.loads(fixed)
        return fixed
    except json.JSONDecodeError:
        return None


def _classify_syntax_error(msg, text):
    """对 JSONDecodeError 消息分类。"""
    msg_lower = msg.lower()

    if 'unexpected EOF' in msg_lower or 'unterminated' in msg_lower:
        return ERROR_TRUNCATION
    if 'expect' in msg_lower and 'property name' in msg_lower:
        return ERROR_MISSING_FIELD
    if 'trailing' in msg_lower:
        return ERROR_SYNTAX
    if 'invalid' in msg_lower and 'control character' in msg_lower:
        return ERROR_SYNTAX
    return ERROR_SYNTAX


def _compute_score(errors, parsed, raw_text):
    """计算格式合规性评分 (0-100)。

    扣分规则：
    - 语法错误: -40（致命，必须重试）
    - 截断: -50（致命，输出不完整）
    - 自动修复后成功: -20（能修复但说明输出质量不高）
    - 空输出: -100（完全无用）
    """
    if parsed is not None and not errors:
        return 100

    if not errors:
        return 100

    score = 100
    for err in errors:
        t = err["type"]
        if t == ERROR_SYNTAX:
            score -= 40
        elif t == ERROR_TRUNCATION:
            score -= 50
        elif t == ERROR_MISSING_FIELD:
            score -= 30
        elif t == "repaired":
            score -= 20
        elif t == ERROR_EMPTY_FIELD:
            score -= 15

    return max(0, score)


def validate_with_schema(text, schema=None):
    """JSON 语法 + Schema 双层校验。

    schema: 可选的自定义校验函数 schema(parsed) -> [(field, type, detail)]
            或内置 check 列表 dict 格式：
            [{"field": "...", "check": callable, "desc": "..."}]
            或 None（只做语法校验）
    """
    score, errors, parsed = validate_json_syntax(text)

    if parsed is None:
        return score, errors, parsed

    if schema is None:
        return score, errors, parsed

    # Schema 校验
    import collections
    schema_errors = []
    if callable(schema):
        raw_errors = schema(parsed)
        if isinstance(raw_errors, collections.abc.Iterable):
            for se in raw_errors:
                field, err_type, detail = se[:3]
                schema_errors.append({
                    "type": err_type or ERROR_MISSING_FIELD,
                    "detail": detail,
                    "line": 0,
                })
    elif isinstance(schema, (list, tuple)):
        for rule in schema:
            field = rule.get("field", "?")
            check = rule.get("check")
            desc = rule.get("desc", "")
            if check and not check(parsed):
                errors.append({
                    "type": rule.get("severity", ERROR_MISSING_FIELD),
                    "detail": f"[{field}] {desc}",
                    "line": 0,
                })

    # 重算分数
    score = _compute_score(errors, parsed, text)
    return score, errors, parsed


# ═══════════════════════════════════════════════════════════════
# 通用于 decisions.json 的 Schema 校验规则
# ═══════════════════════════════════════════════════════════════

# 股票分析结果的 Schema 定义
STOCK_ANALYSIS_SCHEMA = [
    {
        "field": "code",
        "check": lambda d: bool(d.get("code")),
        "desc": "股票代码必须存在",
        "severity": "missing_field",
    },
    {
        "field": "action",
        "check": lambda d: bool(d.get("action")),
        "desc": "操作建议(action)必须存在",
        "severity": "missing_field",
    },
    {
        "field": "stop_loss",
        "check": lambda d: (d.get("stop_loss") or 0) > 0,
        "desc": "止损(stop_loss)必须大于0",
        "severity": "missing_field",
    },
    {
        "field": "take_profit",
        "check": lambda d: (d.get("take_profit") or 0) > 0,
        "desc": "止盈(take_profit)必须存在且大于0",
        "severity": "missing_field",
    },
    {
        "field": "entry_low",
        "check": lambda d: (d.get("entry_low") or 0) > 0,
        "desc": "买入区下沿(entry_low)必须大于0",
        "severity": "missing_field",
    },
    {
        "field": "entry_high",
        "check": lambda d: (d.get("entry_high") or 0) > 0,
        "desc": "买入区上沿(entry_high)必须大于0",
        "severity": "missing_field",
    },
    {
        "field": "entry_range",
        "check": lambda d: (d.get("entry_low") or 0) < (d.get("entry_high") or 0),
        "desc": "买入区下沿 < 上沿",
        "severity": "syntax_error",
    },
    {
        "field": "rr_ratio",
        "check": lambda d: (d.get("rr_ratio") or 0) >= 0,
        "desc": "盈亏比(rr_ratio)必须非负",
        "severity": "missing_field",
    },
    {
        "field": "timing_signal",
        "check": lambda d: bool(d.get("timing_signal")),
        "desc": "时机信号(timing_signal)必须存在",
        "severity": "missing_field",
    },
]

# 用于 <structured_data> 的 Schema
STRUCTURED_DATA_SCHEMA = [
    {
        "field": "holdings",
        "check": lambda d: isinstance(d.get("holdings"), list),
        "desc": "holdings 必须是数组",
        "severity": "missing_field",
    },
    {
        "field": "type",
        "check": lambda d: bool(d.get("type")),
        "desc": "type 字段必须存在",
        "severity": "missing_field",
    },
]


# ═══════════════════════════════════════════════════════════════
# Score-and-Retry 循环
# ═══════════════════════════════════════════════════════════════

def score_and_retry(llm_generate_fn, max_retries=3, threshold=90,
                    schema=None, feedback_prefix="[Format Feedback]",
                    on_retry=None, on_failure=None):
    """通用打分-重试循环包装器。

    Args:
        llm_generate_fn: 调用函数，接收 attempt + previous_errors 字典
                         签名: llm_generate_fn(attempt, context)
                         返回生成的文本
        max_retries: 最大重试次数（含首次，即最多调用 max_retries+1 次）
        threshold: 合格分数线 (0-100)
        schema: 可选 Schema 校验规则
        feedback_prefix: 错误反馈前缀
        on_retry: 可选回调 on_retry(attempt, score, errors)
        on_failure: 可选回调 on_failure(attempts, best_raw, all_errors)

    Returns:
        (parsed, score, errors, raw)
    """
    best_score = 0
    best_raw = None
    best_parsed = None
    all_errors = []

    for attempt in range(max_retries + 1):
        context = {
            "attempt": attempt,
            "previous_errors": all_errors[-1] if all_errors else [],
            "feedback": "",
        }

        # 构建带反馈的调用
        if attempt > 0 and all_errors:
            feedback = _build_feedback(all_errors[-1], feedback_prefix)
            context["feedback"] = feedback

        raw = llm_generate_fn(attempt, context)
        score, errors, parsed = validate_with_schema(raw, schema=schema)

        all_errors.append(errors)

        if on_retry:
            on_retry(attempt, score, errors)

        # 保留最佳结果
        if score > best_score:
            best_score = score
            best_raw = raw
            best_parsed = parsed

        if score >= threshold and parsed is not None:
            # 验收通过
            _record_format_success(attempt, score)
            return parsed, score, errors, raw

        if attempt < max_retries:
            # 记录失败（用于重试）
            _record_format_error(raw, errors, attempt)

    # 所有重试用尽
    if on_failure:
        on_failure(max_retries, best_raw, all_errors)

    _record_format_failure(best_raw, all_errors, max_retries)

    # 返回 best_effort
    if best_parsed is not None:
        return best_parsed, best_score, all_errors[-1], best_raw

    return None, best_score, all_errors[-1], best_raw


def _build_feedback(errors, prefix):
    """将错误列表格式化为模型可读的反馈文本。"""
    parts = [f"{prefix} 上一轮输出存在以下格式问题，请在本次修正："]

    for err in errors:
        t = err.get("type", "unknown")
        d = err.get("detail", "")
        parts.append(f"  - [{t}] {d[:150]}")

    if not errors:
        parts.append("  - 通用格式要求：输出必须是合法的 JSON")

    return "\n".join(parts)


# ═══════════════════════════════════════════════════════════════
# Format Error Library
# ═══════════════════════════════════════════════════════════════

def _ensure_library_path():
    FORMAT_ERROR_LIBRARY_PATH.parent.mkdir(parents=True, exist_ok=True)


def _load_library():
    _ensure_library_path()
    if FORMAT_ERROR_LIBRARY_PATH.exists():
        try:
            return json.loads(FORMAT_ERROR_LIBRARY_PATH.read_text())
        except (json.JSONDecodeError, Exception):
            pass
    return {
        "records": [],
        "stats": {},
        "created_at": datetime.now().isoformat(),
        "last_updated": datetime.now().isoformat(),
    }


def _save_library(lib):
    lib["last_updated"] = datetime.now().isoformat()
    FORMAT_ERROR_LIBRARY_PATH.write_text(
        json.dumps(lib, ensure_ascii=False, indent=2)
    )


def _record_format_error(raw, errors, attempt):
    """记录一次格式错误到档案库。"""
    lib = _load_library()

    record = {
        "timestamp": datetime.now().isoformat(),
        "attempt": attempt + 1,
        "raw_preview": raw[:500] if raw else "",
        "errors": errors,
        "error_types": [e.get("type", "unknown") for e in errors],
    }
    lib["records"].append(record)

    # 更新统计
    for err in errors:
        t = err.get("type", "unknown")
        # 按天统计
        today = datetime.now().strftime("%Y-%m-%d")
        key = f"{today}:{t}"
        lib.setdefault("stats", {}).setdefault("daily", {}).setdefault(key, 0)
        lib["stats"]["daily"][key] = lib["stats"]["daily"].get(key, 0) + 1

    # 容量控制：保留最近 1000 条
    if len(lib["records"]) > 1000:
        lib["records"] = lib["records"][-1000:]

    _save_library(lib)


def _record_format_success(attempt, score):
    """记录一次成功的格式校验。"""
    lib = _load_library()
    today = datetime.now().strftime("%Y-%m-%d")
    key = f"{today}:success"
    lib.setdefault("stats", {}).setdefault("daily", {}).setdefault(key, 0)
    lib["stats"]["daily"][key] = lib["stats"]["daily"].get(key, 0) + 1
    lib.setdefault("stats", {}).setdefault("total_calls", 0)
    lib["stats"]["total_calls"] = lib["stats"].get("total_calls", 0) + 1
    _save_library(lib)


def _record_format_failure(raw, all_errors, max_retries):
    """记录超出最大重试次数的完全失败。"""
    lib = _load_library()
    today = datetime.now().strftime("%Y-%m-%d")
    key = f"{today}:total_failure"
    lib.setdefault("stats", {}).setdefault("daily", {}).setdefault(key, 0)
    lib["stats"]["daily"][key] = lib["stats"]["daily"].get(key, 0) + 1
    lib.setdefault("stats", {}).setdefault("total_calls", 0)
    lib["stats"]["total_calls"] = lib["stats"].get("total_calls", 0) + 1
    lib["stats"]["total_failures"] = lib["stats"].get("total_failures", 0) + 1
    _save_library(lib)


# ═══════════════════════════════════════════════════════════════
# 统计与报告
# ═══════════════════════════════════════════════════════════════

def get_error_report(days=7):
    """生成格式错误统计报告。"""
    lib = _load_library()

    daily = lib.get("stats", {}).get("daily", {})
    total_calls = lib.get("stats", {}).get("total_calls", 0)
    total_failures = lib.get("stats", {}).get("total_failures", 0)

    # 按错误类型聚合（近 N 天）
    type_count = Counter()
    today = datetime.now()
    for key, count in daily.items():
        try:
            date_str, err_type = key.split(":", 1)
            date = datetime.strptime(date_str, "%Y-%m-%d")
            if (today - date).days <= days:
                type_count[err_type] += count
        except (ValueError, IndexError):
            pass

    # 总调用数
    total_typed = sum(type_count.values())
    success_count = type_count.get("success", 0)

    report = {
        "period_days": days,
        "total_calls": total_calls,
        "total_failures": total_failures,
        "period_calls": total_typed,
        "period_success": success_count,
        "period_error_rate": (
            round((1 - success_count / max(total_typed, 1)) * 100, 1)
            if total_typed > 0 else 0
        ),
        "error_breakdown": dict(type_count.most_common()),
        "recent_records": lib["records"][-20:],
    }

    return report


def get_auto_fix_recommendations():
    """根据错误统计自动生成修复建议。

    规则：
    - missing_field 连续 7 天 > 5 次/天 → 检查 prompt 字段排序
    - syntax_error 连续 7 天 > 3 次/天 → 建议启用 response_format
    - truncation > 3 次/周 → 检查 max_tokens
    """
    lib = _load_library()
    daily = lib.get("stats", {}).get("daily", {})

    # 近 7 天统计
    today = datetime.now()
    week_errors = defaultdict(int)
    for key, count in daily.items():
        try:
            date_str, err_type = key.split(":", 1)
            date = datetime.strptime(date_str, "%Y-%m-%d")
            if (today - date).days <= 7:
                week_errors[err_type] += count
        except (ValueError, IndexError):
            pass

    recommendations = []

    if week_errors.get("missing_field", 0) > 35:  # 5次/天 * 7天
        recommendations.append(
            "missing_field 频繁（近7天{}次）→ 检查 prompt 中字段是否过于靠后，"
            "考虑将 schema 定义提前到 prompt 前部".format(
                week_errors.get("missing_field", 0)
            )
        )

    if week_errors.get("syntax_error", 0) > 21:  # 3次/天 * 7天
        recommendations.append(
            "syntax_error 频繁（近7天{}次）→ 建议启用 response_format=json_object "
            "或 constrained decoding".format(
                week_errors.get("syntax_error", 0)
            )
        )

    if week_errors.get("truncation", 0) > 3:
        recommendations.append(
            "truncation {}次/周 → 检查 max_tokens 是否足够，建议调高 20%".format(
                week_errors.get("truncation", 0)
            )
        )

    total_errors = sum(v for k, v in week_errors.items() if k != "success")
    total_calls = week_errors.get("success", 0) + total_errors
    error_rate = total_errors / max(total_calls, 1) * 100

    if error_rate > 5 and total_calls > 20:
        recommendations.append(
            "总错误率 {:.1f}%（{}次/{}次）→ 循环软约束已达极限，建议接入 "
            "response_format=json_object 或 outlines JSON Schema generation".format(
                error_rate, total_errors, total_calls
            )
        )

    return recommendations


# ═══════════════════════════════════════════════════════════════
# 便捷包装器 — 用于 update_data.py 等下游
# ═══════════════════════════════════════════════════════════════

def safe_parse_structured(text, schema=None, log_errors=True):
    """安全解析 <structured_data> JSON，带评分重试语义。

    用于 update_data.py 的 parse_report() 替代 json.loads()。

    Args:
        text: markdown 文本
        schema: 可选 Schema
        log_errors: 是否记录到 FormatErrorLibrary

    Returns:
        (parsed, score, errors)
        parsed 为 None 时代表完全失败
    """
    score, errors, parsed = validate_with_schema(text, schema=schema)

    if log_errors and errors and parsed is None:
        _record_format_error(text, errors, 0)

    return parsed, score, errors


# ═══════════════════════════════════════════════════════════════
# CLI 入口 — 可直接作为工具使用
# ═══════════════════════════════════════════════════════════════

def main():
    """CLI: 校验从 stdin 或文件传入的 JSON。"""
    import argparse

    parser = argparse.ArgumentParser(
        description="JSON 格式校验器 — 知微股票分析 Pipeline"
    )
    parser.add_argument("file", nargs="?", help="要校验的文件路径（默认从 stdin 读）")
    parser.add_argument("--schema", choices=["decisions", "structured", "none"],
                        default="none", help="Schema 校验模式")
    parser.add_argument("--report", action="store_true",
                        help="生成格式错误统计报告")
    parser.add_argument("--recommend", action="store_true",
                        help="生成自动修复建议")
    parser.add_argument("--days", type=int, default=7,
                        help="统计报告的天数范围（默认 7）")

    args = parser.parse_args()

    if args.report:
        report = get_error_report(days=args.days)
        print(json.dumps(report, ensure_ascii=False, indent=2))
        return

    if args.recommend:
        recs = get_auto_fix_recommendations()
        if recs:
            print("## 自动修复建议\n")
            for r in recs:
                print(f"- {r}")
        else:
            print("无待修复问题。格式错误率在可接受范围内。")
        return

    # 读取输入
    if args.file:
        with open(args.file, "r", encoding="utf-8") as f:
            text = f.read()
    else:
        text = sys.stdin.read()

    # 选择 Schema
    schema = None
    if args.schema == "decisions":
        schema = STOCK_ANALYSIS_SCHEMA
    elif args.schema == "structured":
        schema = STRUCTURED_DATA_SCHEMA

    score, errors, parsed = validate_with_schema(text, schema=schema)

    result = {
        "score": score,
        "passed": score >= 90,
        "errors": errors,
        "parsed_preview": str(parsed)[:300] if parsed else None,
    }

    print(json.dumps(result, ensure_ascii=False, indent=2))

    if errors:
        sys.exit(1)


if __name__ == "__main__":
    main()