【案例】金融行业动态方向评估（二）：信息抽取与文本匹配

# -*- coding: utf-8 -*-
"""把模型吐出来的那段文本尽量变成 dict。

模型返回的永远是字符串。它可能规规矩矩给 JSON，也可能：
  · 套一层 ```json ... ``` 围栏
  · 用中文引号、中文逗号
  · 用 Python 风格的单引号
  · 在 JSON 前后再写两句寒暄
  · 结尾多一个逗号
这个文件只做一件事：尽量救回来；救不回来就返回 None，绝不返回半成品。

纯标准库，`python3 json_repair.py` 直接跑自测。
"""
import json
import re

FENCE_RE = re.compile(r"```(?:json|JSON)?\s*(.*?)```", re.DOTALL)
OBJ_RE = re.compile(r"\{.*\}", re.DOTALL)
TRAILING_COMMA_RE = re.compile(r",\s*([}\]])")

# 全角标点 → 半角，只在结构层面替换，值里的中文不受影响（见下方 _protect）
PUNCT_MAP = {
    "“": '"', "”": '"', "‘": "'", "’": "'",
    "：": ":", "，": ",", "【": "[", "】": "]",
}


def strip_fence(text: str) -> str:
    """去掉 Markdown 代码围栏，只保留里面那段。"""
    found = FENCE_RE.findall(text)
    for block in found:
        if block.strip():
            return block.strip()
    return text.strip()


def slice_object(text: str) -> str:
    """把首尾的寒暄砍掉，只留最外层大括号那一段。"""
    match = OBJ_RE.search(text)
    return match.group(0) if match else text


def normalize_punct(text: str) -> str:
    """把结构性的全角标点换成半角。"""
    for src, dst in PUNCT_MAP.items():
        text = text.replace(src, dst)
    return text


def parse_model_json(text: str):
    """主入口：返回 dict，救不回来返回 None。

    逐级放宽，每一级都先试一次标准解析，能早退就早退。
    """
    if not isinstance(text, str) or not text.strip():
        return None

    candidates = []
    stage1 = strip_fence(text)
    candidates.append(stage1)
    stage2 = slice_object(stage1)
    candidates.append(stage2)
    stage3 = normalize_punct(stage2)
    candidates.append(stage3)
    candidates.append(TRAILING_COMMA_RE.sub(r"\1", stage3))
    # 最后一招：Python 风格单引号换成双引号（值里含撇号时会失败，所以放最后）
    candidates.append(TRAILING_COMMA_RE.sub(r"\1", stage3).replace("'", '"'))

    for candidate in candidates:
        try:
            data = json.loads(candidate)
        except (ValueError, TypeError):
            continue
        if isinstance(data, dict):
            return data
    return None


def _demo():
    cases = [
        ('{"日期": ["2023-02-15"]}', True, "标准 JSON"),
        ('```json\n{"日期": ["2023-02-15"]}\n```', True, "带围栏"),
        ("好的，结果如下：\n{\"日期\": [\"2023-02-15\"]}\n希望有帮助。", True, "前后有寒暄"),
        ("{'日期': ['2023-02-15']}", True, "单引号"),
        ('{"日期": ["2023-02-15"],}', True, "尾随逗号"),
        ('{“日期”: [“2023-02-15”]}', True, "全角引号"),
        ("原文中未提及", False, "根本不是 JSON"),
        ('{"日期": ["2023-02-15"', False, "截断了"),
        ("", False, "空字符串"),
    ]
    ok = 0
    for raw, should_parse, label in cases:
        data = parse_model_json(raw)
        got = data is not None
        flag = "通过" if got == should_parse else "不符"
        if got == should_parse:
            ok += 1
        print(f"[{flag}] {label:<12} 解析={'成功' if got else '失败'} -> {data}")
    print(f"\n{ok}/{len(cases)} 条符合预期")
    assert ok == len(cases), "容错解析行为与预期不符"

    # 关键断言：救不回来时必须是 None，不能是字符串、不能是空 dict
    assert parse_model_json("模型今天心情不好") is None
    assert parse_model_json("[1,2,3]") is None, "顶层不是对象时也要判失败"
    print("断言全部通过：解析失败一律返回 None，绝不返回半成品")


if __name__ == "__main__":
    _demo()

# -*- coding: utf-8 -*-
"""抽取结果校验器：解析成功 ≠ 结果可用。

parse_model_json 只保证「它是个 dict」。这个文件保证：
  · schema 里的字段一个不少（少的补成「原文中未提及」）
  · 没有 schema 之外的野字段（模型很爱自己加「涨跌幅」）
  · 每个值都是 list[str]，单个字符串自动包成 list
  · 「原文中未提及」必须独占，不能和真值混在同一个 list 里

纯标准库，`python3 ie_schema_check.py` 直接跑自测。
"""
NOT_MENTIONED = "原文中未提及"


def normalize(data: dict, fields: list, placeholder: str = NOT_MENTIONED) -> dict:
    """把模型给的 dict 规整成「字段齐全、值都是 list[str]」的标准形状。

    只做形状上的修复，不猜内容：缺失字段一律补 placeholder。
    """
    fixed = {}
    for field in fields:
        value = data.get(field, None)
        if value is None or value == "" or value == []:
            fixed[field] = [placeholder]
            continue
        if isinstance(value, (str, int, float)):
            value = [str(value)]
        elif isinstance(value, list):
            value = [str(v) for v in value if str(v).strip() != ""]
            if not value:
                value = [placeholder]
        else:
            value = [str(value)]
        fixed[field] = value
    return fixed


def check(data: dict, fields: list, placeholder: str = NOT_MENTIONED) -> list:
    """返回问题清单；空列表表示这条结果可以进下一环节。"""
    problems = []
    if not isinstance(data, dict):
        return ["顶层不是对象"]

    for field in fields:
        if field not in data:
            problems.append(f"缺字段：{field}")
            continue
        value = data[field]
        if not isinstance(value, list):
            problems.append(f"{field} 的值不是列表：{type(value).__name__}")
            continue
        if not value:
            problems.append(f"{field} 是空列表，应写成 ['{placeholder}']")
            continue
        if any(not isinstance(v, str) for v in value):
            problems.append(f"{field} 里有非字符串元素")
        if placeholder in value and len(value) > 1:
            problems.append(f"{field} 把占位符和真值混在了一起：{value}")

    extra = [k for k in data.keys() if k not in fields]
    if extra:
        problems.append(f"多出 schema 之外的字段：{extra}")
    return problems


def _demo():
    fields = ["日期", "股票名称", "开盘价", "收盘价", "成交量"]

    # 一条典型的「模型自由发挥」结果：漏了收盘价、成交量写成了数字、还自己加了一个字段
    raw = {
        "日期": ["2023-02-15"],
        "股票名称": "佰笃[BD]美股",
        "开盘价": ["10美元"],
        "成交量": 460000,
        "投资者情绪": ["平稳"],
    }

    print("原始：", raw)
    problems = check(raw, fields)
    print("\n校验问题：")
    for p in problems:
        print("  ·", p)
    assert problems, "这条明显有问题，校验器却没报"

    fixed = normalize(raw, fields)
    print("\n规整后：")
    for k, v in fixed.items():
        print(f"  {k}: {v}")

    problems_after = check(fixed, fields)
    print("\n规整后的校验问题：", problems_after or "无")
    assert problems_after == [], "规整后仍不合格，说明 normalize 漏了情况"

    # normalize 只修形状，不编内容：收盘价原文里没有，就必须是占位符
    assert fixed["收盘价"] == [NOT_MENTIONED]
    # 野字段在 normalize 后被丢掉（schema 说了算）
    assert "投资者情绪" not in fixed
    # 数字被转成字符串，不做单位猜测
    assert fixed["成交量"] == ["460000"]

    # 占位符与真值混用必须被拦住
    bad = dict(fixed)
    bad["开盘价"] = ["10美元", NOT_MENTIONED]
    assert any("混在了一起" in p for p in check(bad, fields))

    print("\n断言全部通过：normalize 只修形状，check 只报问题，两者都不编内容")


if __name__ == "__main__":
    _demo()

# -*- coding: utf-8 -*-
"""字段级评分器：整条对错会把信息压没，字段级才看得见问题在哪。

两种口径：
  · 整条准确率（exact match）：五个字段全对才算一条对
  · 字段级准确率（field-level）：逐字段比，一条错一个字段只扣一个字段的分
再按字段拆开，能直接指出「哪个字段在拖后腿」——这是改提示词的依据。

纯标准库，`python3 ie_field_score.py` 直接跑。
"""
NOT_MENTIONED = "原文中未提及"

# 评测集：人工标注的答案（gold）与模型给出的结果（pred）
GOLD = [
    {
        "日期": ["2023-02-15"], "股票名称": ["佰笃[BD]美股"],
        "开盘价": ["10美元"], "收盘价": ["13美元"], "成交量": ["460,000"],
    },
    {
        "日期": ["2023-04-05"], "股票名称": ["盘古(0021)"],
        "开盘价": ["23元"], "收盘价": ["26美元"], "成交量": ["310,000"],
    },
    {
        "日期": ["2023-05-20"], "股票名称": ["长虹科技"],
        "开盘价": ["48元"], "收盘价": [NOT_MENTIONED], "成交量": ["120,000"],
    },
]

PRED = [
    {   # 成交量丢了千分位
        "日期": ["2023-02-15"], "股票名称": ["佰笃[BD]美股"],
        "开盘价": ["10美元"], "收盘价": ["13美元"], "成交量": ["460000"],
    },
    {   # 股票名称把括号里的代码扔了，成交量同样丢千分位
        "日期": ["2023-04-05"], "股票名称": ["盘古"],
        "开盘价": ["23元"], "收盘价": ["26美元"], "成交量": ["310000"],
    },
    {   # 原文没有收盘价，模型自己编了一个——最危险的一类错
        "日期": ["2023-05-20"], "股票名称": ["长虹科技"],
        "开盘价": ["48元"], "收盘价": ["51元"], "成交量": ["120,000"],
    },
]

FIELDS = ["日期", "股票名称", "开盘价", "收盘价", "成交量"]


def norm_value(values: list) -> tuple:
    """比较前的归一化：去空白、去千分位逗号。顺序不敏感，所以排序成元组。"""
    cleaned = [str(v).strip().replace(",", "").replace("，", "") for v in values]
    return tuple(sorted(cleaned))


def score(gold_list: list, pred_list: list, fields: list) -> dict:
    assert len(gold_list) == len(pred_list), "评测集与预测条数必须一一对应"
    per_field = {f: {"right": 0, "total": 0} for f in fields}
    exact_right = 0
    hallucinated = 0     # 金标是「未提及」，模型却给了具体值
    missed = 0           # 金标有值，模型答「未提及」

    for gold, pred in zip(gold_list, pred_list):
        all_right = True
        for field in fields:
            g = norm_value(gold.get(field, [NOT_MENTIONED]))
            p = norm_value(pred.get(field, [NOT_MENTIONED]))
            per_field[field]["total"] += 1
            if g == p:
                per_field[field]["right"] += 1
            else:
                all_right = False
                if g == (NOT_MENTIONED,) and p != (NOT_MENTIONED,):
                    hallucinated += 1
                if g != (NOT_MENTIONED,) and p == (NOT_MENTIONED,):
                    missed += 1
        if all_right:
            exact_right += 1

    field_right = sum(v["right"] for v in per_field.values())
    field_total = sum(v["total"] for v in per_field.values())
    return {
        "exact": exact_right / len(gold_list),
        "field": field_right / field_total,
        "per_field": per_field,
        "hallucinated": hallucinated,
        "missed": missed,
    }


def _report(result: dict):
    print(f"整条准确率 exact match : {result['exact']:.3f}")
    print(f"字段级准确率 field-level: {result['field']:.3f}")
    print("\n逐字段：")
    for field, stat in result["per_field"].items():
        acc = stat["right"] / stat["total"]
        bar = "█" * int(acc * 20)
        print(f"  {field:<6} {stat['right']}/{stat['total']}  {acc:.2f} {bar}")
    print(f"\n凭空编出来的值：{result['hallucinated']} 个")
    print(f"该抽没抽到的值：{result['missed']} 个")


if __name__ == "__main__":
    result = score(GOLD, PRED, FIELDS)
    _report(result)

    # 三条里只有第一条全对（千分位归一化后成交量算对）→ exact = 1/3
    assert abs(result["exact"] - 1 / 3) < 1e-9
    # 错的只有两个字段：第二条的股票名称、第三条的收盘价 → 13/15
    assert abs(result["field"] - 13 / 15) < 1e-9
    # 只看 exact，这一版像是「三条错两条」；字段级立刻指出 86.7% 的字段是对的
    assert result["field"] > result["exact"]
    assert result["per_field"]["成交量"]["right"] == 3, "千分位已归一化，应判对"
    assert result["per_field"]["日期"]["right"] == 3
    assert result["per_field"]["股票名称"]["right"] == 2
    assert result["per_field"]["收盘价"]["right"] == 2
    # 第三条原文没有收盘价，模型编了个 51 元——这类错必须单独计数
    assert result["hallucinated"] == 1 and result["missed"] == 0
    print("\n断言全部通过：同一份结果，两种口径差出 0.333 与 0.867")

# -*- coding: utf-8 -*-
"""金融文本信息抽取：把一句话抽成固定字段的 JSON。

运行前提：本机已装 Ollama 并拉好模型（ollama run qwen3:8b），
        以及 Python 包 ollama 与 rich。
环境变量：
    OLLAMA_MODEL  模型名，默认 qwen3:8b
    OLLAMA_HOST   服务地址，默认走 ollama 库自己的 http://127.0.0.1:11434
"""
import json
import os
import sys

import ollama
from rich import print
from rich.console import Console

# 1. schema：这张表单上有哪几个格子，一次定死，全流程复用
SCHEMA = {
    "金融": ["日期", "股票名称", "开盘价", "收盘价", "成交量"],
}

# 抽不到时统一填这个值，而不是留空、不是 null、不是漏键
NOT_MENTIONED = "原文中未提及"

# 2. 提示词骨架：句子在前，要求在后，最后一句把输出格式钉死
IE_PATTERN = (
    "{}\n\n提取上述句子中{}的实体，并按照JSON格式输出，"
    "上述句子中不存在的信息用['" + NOT_MENTIONED + "']来表示，多个值之间用','分隔。"
)

# 3. few-shot：给一问一答，让模型看见「答案长什么样」
IE_EXAMPLES = {
    "金融": [
        {
            "content": "2023-01-10，股市震荡。股票古哥-D[EOOE]美股今日开盘价100美元，"
                       "一度飙升至105美元，随后回落至98美元，最终以102美元收盘，成交量达到520000。",
            "answers": {
                "日期": ["2023-01-10"],
                "股票名称": ["古哥-D[EOOE]美股"],
                "开盘价": ["100美元"],
                "收盘价": ["102美元"],
                "成交量": ["520000"],
            },
        }
    ]
}


def build_prompt(sentence: str, entity_type: str) -> str:
    """把一句话套进 IE_PATTERN，得到最终发给模型的那段文本。"""
    fields = ", ".join(SCHEMA[entity_type])
    return IE_PATTERN.format(sentence, f"“{entity_type}”({fields})")


def init_prompts() -> list:
    """构造前置消息：system + 若干组 few-shot 问答。"""
    history = [{"role": "system", "content": "你是一个信息抽取助手。"}]
    for entity_type, examples in IE_EXAMPLES.items():
        for example in examples:
            history.append({
                "role": "user",
                "content": build_prompt(example["content"], entity_type),
            })
            history.append({
                "role": "assistant",
                "content": json.dumps(example["answers"], ensure_ascii=False),
            })
    return history


def call_model(messages: list, model: str) -> str:
    """调一次模型，只负责拿回字符串；任何异常都往上抛，由调用方决定重试。"""
    resp = ollama.chat(model=model, messages=messages)
    return resp["message"]["content"]


def extract_once(sentence: str, history: list, model: str) -> str:
    """单次抽取，返回模型的原始文本输出。"""
    messages = list(history) + [
        {"role": "user", "content": build_prompt(sentence, "金融")}
    ]
    return call_model(messages, model)


def extract(sentence: str, history: list, model: str, retries: int = 2) -> dict:
    """带重试的抽取：解析失败就重来，重来时把错误回喂给模型。

    返回值一定是 dict；连续失败则抛 ValueError，由上层决定是丢弃还是转人工。
    """
    from json_repair import parse_model_json  # 解析容错逻辑单独成文件，便于复用与单测
    from ie_schema_check import normalize, check

    last_raw = ""
    for attempt in range(retries + 1):
        messages = list(history)
        messages.append({"role": "user", "content": build_prompt(sentence, "金融")})
        if attempt > 0:
            messages.append({"role": "assistant", "content": last_raw})
            messages.append({
                "role": "user",
                "content": "上一条回复不是合法 JSON。请只输出一个 JSON 对象，"
                           "不要写解释、不要加代码围栏。",
            })
        last_raw = call_model(messages, model)
        data = parse_model_json(last_raw)
        if data is None:
            continue
        data = normalize(data, SCHEMA["金融"], NOT_MENTIONED)
        problems = check(data, SCHEMA["金融"])
        if not problems:
            return data
    raise ValueError(f"连续 {retries + 1} 次未拿到合法结果，最后一次原文：{last_raw[:200]}")


def main() -> int:
    console = Console()
    model = os.environ.get("OLLAMA_MODEL", "qwen3:8b")
    history = init_prompts()

    sentences = [
        "2023-02-15，寓意吉祥的节日，股票佰笃[BD]美股开盘价10美元，虽然经历了波动，"
        "但最终以13美元收盘，成交量微幅增加至460,000，投资者情绪较为平稳。",
        "2023-04-05，市场迎来轻松氛围，股票盘古(0021)开盘价23元，尽管经历了波动，"
        "但最终以26美元收盘，成交量缩小至310,000，投资者保持观望态度。",
    ]

    failed = 0
    for sentence in sentences:
        with console.status("[bold bright_green]正在抽取..."):
            try:
                result = extract(sentence, history, model)
            except ollama.ResponseError as exc:      # 模型没拉下来、服务没起
                print(f"[bold red]模型调用失败：{exc}")
                return 2
            except ValueError as exc:                # 输出始终解析不了
                print(f"[bold yellow]放弃这一条：{exc}")
                failed += 1
                continue
        print(f"[bold bright_red]原文：{sentence}")
        print(f"[bold bright_green]抽取：{json.dumps(result, ensure_ascii=False)}")

    print(f"[bold]共 {len(sentences)} 条，放弃 {failed} 条")
    return 0


if __name__ == "__main__":
    sys.exit(main())

# -*- coding: utf-8 -*-
"""金融文本匹配：判断两段文本是不是在说同一件事。

运行前提：本机已装 Ollama 并拉好模型（ollama run qwen3:8b），
        以及 Python 包 ollama 与 rich。
环境变量：
    OLLAMA_MODEL  模型名，默认 qwen3:8b
"""
import os
import sys

import ollama
from rich import print
from rich.console import Console

POSITIVE = "是"
NEGATIVE = "不是"

SYSTEM_PROMPT = (
    "现在你需要帮助我完成文本匹配任务，当我给你两个句子时，"
    "你需要回答我这两句话语义是否相似。只需要回答是否相似，不要做多余的回答。"
)

# few-shot：正负样本都要给，而且负样本要给「像但不是」的那种
EXAMPLES = {
    POSITIVE: [
        ("公司ABC发布了季度财报，显示盈利增长。", "财报披露，公司ABC利润上升。"),
    ],
    NEGATIVE: [
        ("黄金价格下跌，投资者抛售。", "外汇市场交易额创下新高。"),
        ("央行降息，刺激经济增长。", "新能源技术的创新。"),
    ],
}

PAIR_PATTERN = "句子一: {}\n句子二: {}\n上面两句话是相似的语义吗？"


def init_prompts() -> list:
    """system + few-shot 一问一答，构成前置消息。"""
    history = [{"role": "system", "content": SYSTEM_PROMPT}]
    for label, pairs in EXAMPLES.items():
        for s1, s2 in pairs:
            history.append({"role": "user", "content": PAIR_PATTERN.format(s1, s2)})
            history.append({"role": "assistant", "content": label})
    return history


def normalize_answer(text: str):
    """把模型的自由回答收敛成 是 / 不是 / None。

    「不是」必须先判，否则「不是」会被「是」的子串匹配吃掉——
    这是本任务最容易写反的一行。
    """
    if not isinstance(text, str):
        return None
    cleaned = text.strip().replace(" ", "").replace("\n", "")
    for marker in ("不是", "不相似", "否"):
        if marker in cleaned:
            return NEGATIVE
    for marker in ("是", "相似"):
        if marker in cleaned:
            return POSITIVE
    return None


def match_pair(s1: str, s2: str, history: list, model: str, retries: int = 1):
    """返回 是 / 不是；模型答非所问就重试，仍失败返回 None 交人工。"""
    for attempt in range(retries + 1):
        messages = list(history)
        messages.append({"role": "user", "content": PAIR_PATTERN.format(s1, s2)})
        if attempt > 0:
            messages.append({
                "role": "user",
                "content": f"请只回答「{POSITIVE}」或「{NEGATIVE}」两个词之一。",
            })
        raw = ollama.chat(model=model, messages=messages)["message"]["content"]
        answer = normalize_answer(raw)
        if answer is not None:
            return answer, raw
    return None, raw


def main() -> int:
    console = Console()
    model = os.environ.get("OLLAMA_MODEL", "qwen3:8b")
    history = init_prompts()

    sentence_pairs = [
        ("股票市场今日大涨，投资者乐观。", "持续上涨的市场让投资者感到满意。"),
        ("油价大幅下跌，能源公司面临挑战。", "未来智能城市的建设趋势愈发明显。"),
        ("利率上升，影响房地产市场。", "高利率对房地产有一定冲击。"),
    ]

    undecided = 0
    for s1, s2 in sentence_pairs:
        with console.status("[bold bright_green]正在判定..."):
            try:
                answer, raw = match_pair(s1, s2, history, model)
            except ollama.ResponseError as exc:
                print(f"[bold red]模型调用失败：{exc}")
                return 2
        print(f"[bold bright_red]句子一：{s1}")
        print(f"[bold bright_red]句子二：{s2}")
        if answer is None:
            undecided += 1
            print(f"[bold yellow]判不了，原始回复：{raw[:80]}")
        else:
            print(f"[bold bright_green]判定：{answer}")

    print(f"[bold]共 {len(sentence_pairs)} 对，判不了 {undecided} 对")
    return 0


if __name__ == "__main__":
    sys.exit(main())

# -*- coding: utf-8 -*-
"""文本匹配的评测指标：准确率单独看会骗人。

同一批预测，在不同正负比例的评测集上算出来的准确率完全不同。
这个文件把混淆矩阵、精确率、召回率、F1 一次算清楚，并用一个
「全判不是」的退化模型证明：准确率 0.9 也可能一无是处。

纯标准库，`python3 match_metrics.py` 直接跑。
"""
POSITIVE = "是"
NEGATIVE = "不是"


def confusion(gold: list, pred: list) -> dict:
    """返回四格计数。约定：正类 = 「是」（两句话指向同一件事）。"""
    assert len(gold) == len(pred), "金标与预测条数必须一致"
    tp = fp = fn = tn = 0
    for g, p in zip(gold, pred):
        if g == POSITIVE and p == POSITIVE:
            tp += 1
        elif g == NEGATIVE and p == POSITIVE:
            fp += 1
        elif g == POSITIVE and p == NEGATIVE:
            fn += 1
        else:
            tn += 1
    return {"TP": tp, "FP": fp, "FN": fn, "TN": tn}


def metrics(cm: dict) -> dict:
    tp, fp, fn, tn = cm["TP"], cm["FP"], cm["FN"], cm["TN"]
    total = tp + fp + fn + tn
    acc = (tp + tn) / total if total else 0.0
    precision = tp / (tp + fp) if (tp + fp) else 0.0
    recall = tp / (tp + fn) if (tp + fn) else 0.0
    f1 = 2 * precision * recall / (precision + recall) if (precision + recall) else 0.0
    return {"准确率": acc, "精确率": precision, "召回率": recall, "F1": f1}


def show(title: str, gold: list, pred: list):
    cm = confusion(gold, pred)
    m = metrics(cm)
    print(f"\n{title}")
    print(f"  样本 {len(gold)} 条，其中正类 {gold.count(POSITIVE)} 条")
    print(f"  TP={cm['TP']}  FP={cm['FP']}  FN={cm['FN']}  TN={cm['TN']}")
    print("  " + "  ".join(f"{k}={v:.3f}" for k, v in m.items()))
    return cm, m


if __name__ == "__main__":
    # 评测集 A：正负各半，比较诚实的一份
    gold_a = [POSITIVE] * 5 + [NEGATIVE] * 5
    pred_a = [POSITIVE, POSITIVE, POSITIVE, NEGATIVE, POSITIVE,
              NEGATIVE, NEGATIVE, POSITIVE, NEGATIVE, NEGATIVE]
    cm_a, m_a = show("评测集 A：正负 5:5", gold_a, pred_a)

    # 评测集 B：只有 1 条正类，其余全是负类——真实语料里非常常见
    gold_b = [POSITIVE] + [NEGATIVE] * 9
    pred_b = [NEGATIVE] * 10          # 一个退化模型：闭着眼睛全答「不是」
    cm_b, m_b = show("评测集 B：正负 1:9，模型全答「不是」", gold_b, pred_b)

    print("\n" + "─" * 52)
    print(f"A 的准确率 {m_a['准确率']:.3f}，F1 {m_a['F1']:.3f}")
    print(f"B 的准确率 {m_b['准确率']:.3f}，F1 {m_b['F1']:.3f}")
    print("B 的准确率更高，但它一条正类都没找出来。")

    # 断言：退化模型的准确率高于真模型，F1 却是 0
    assert m_b["准确率"] > m_a["准确率"], "样本失衡下准确率会被负类抬上去"
    assert m_b["F1"] == 0.0 and m_b["召回率"] == 0.0
    assert m_a["F1"] > 0.6

    # 精确率与召回率的方向相反，必须一起看
    assert abs(m_a["精确率"] - 4 / 5) < 1e-9   # 判「是」5 次，对 4 次
    assert abs(m_a["召回率"] - 4 / 5) < 1e-9   # 5 条真正类，找回 4 条

    print("\n断言全部通过：只报准确率的匹配实验，等于没报")

# -*- coding: utf-8 -*-
"""阈值扫描：让模型给分而不是给结论，阈值由评测集定，不由手感定。

只要能拿到一个 0~1 的「相似度分数」，判「是」还是「不是」就是一条线的事。
这条线定在哪，直接决定精确率与召回率往哪边倒。
扫一遍所有候选阈值，挑 F1 最高的那个——这才是「阈值怎么定」的答案。

纯标准库，`python3 match_threshold_scan.py` 直接跑。
"""
from match_metrics import confusion, metrics, POSITIVE, NEGATIVE

# 评测集：(金标, 模型给的相似度分数)
# 分数可以来自「回答『是』时的自报置信度」，也可以来自两句话的向量余弦
SAMPLES = [
    ("是", 0.95), ("是", 0.88), ("是", 0.81), ("是", 0.72), ("是", 0.64),
    ("是", 0.58), ("不是", 0.61), ("不是", 0.55), ("不是", 0.42), ("不是", 0.33),
    ("不是", 0.21), ("不是", 0.12), ("是", 0.49), ("不是", 0.68), ("是", 0.77),
]


def predict(samples: list, threshold: float) -> list:
    """分数 >= 阈值判为正类。边界取闭区间，全流程必须统一。"""
    return [POSITIVE if score >= threshold else NEGATIVE for _, score in samples]


def scan(samples: list, thresholds: list) -> list:
    gold = [label for label, _ in samples]
    rows = []
    for th in thresholds:
        pred = predict(samples, th)
        m = metrics(confusion(gold, pred))
        rows.append((th, m))
    return rows


def _bar(value: float, width: int = 16) -> str:
    return "█" * int(round(value * width))


if __name__ == "__main__":
    thresholds = [round(0.30 + 0.05 * i, 2) for i in range(13)]   # 0.30 ~ 0.90
    rows = scan(SAMPLES, thresholds)

    print(f"{'阈值':<6}{'精确率':<8}{'召回率':<8}{'F1':<8}")
    print("─" * 44)
    best_th, best_f1 = None, -1.0
    for th, m in rows:
        print(f"{th:<8.2f}{m['精确率']:<10.3f}{m['召回率']:<10.3f}"
              f"{m['F1']:<8.3f}{_bar(m['F1'])}")
        if m["F1"] > best_f1:
            best_th, best_f1 = th, m["F1"]

    print("─" * 44)
    print(f"F1 最高的阈值：{best_th:.2f}（F1={best_f1:.3f}）")

    low = dict(rows)[0.30]
    high = dict(rows)[0.90]
    print(f"\n阈值 0.30：召回率 {low['召回率']:.3f}，精确率 {low['精确率']:.3f}"
          f"  —— 宁可错抓，不肯放过")
    print(f"阈值 0.90：召回率 {high['召回率']:.3f}，精确率 {high['精确率']:.3f}"
          f"  —— 宁可放过，不肯错抓")

    # 阈值越高，召回率单调不增；精确率总体不减。这是这张表最该记住的形状
    recalls = [m["召回率"] for _, m in rows]
    assert all(recalls[i] >= recalls[i + 1] for i in range(len(recalls) - 1)), \
        "阈值升高时召回率不可能变大"
    assert low["召回率"] == 1.0, "阈值 0.30 低于所有正类分数，应当全部召回"
    assert high["精确率"] >= low["精确率"]
    assert best_f1 > max(low["F1"], high["F1"]), "最优阈值应当好过两个极端"
    print("\n断言全部通过：阈值是扫出来的，不是拍出来的")

# -*- coding: utf-8 -*-
"""评测集体检：先看这份评测集本身合不合格，再看模型分数。

抽取任务与匹配任务的评测集判据不同，这个文件把两边都查一遍：
  · 抽取：每条都要有全部字段的金标；「未提及」的字段要有一定占比，
          否则「抽不到时返回什么」这条规则根本没被测到
  · 匹配：正负比例不能太偏；负样本里要有「话题相近但不是同一件事」的难例，
          全是「风马牛不相及」的负样本会把分数刷得虚高

纯标准库，`python3 eval_set_build.py` 直接跑。
"""
NOT_MENTIONED = "原文中未提及"
FIELDS = ["日期", "股票名称", "开盘价", "收盘价", "成交量"]


def audit_ie_set(gold_list: list, fields: list) -> dict:
    """抽取评测集体检。"""
    missing_field_rows = []
    placeholder_hits = {f: 0 for f in fields}
    for idx, gold in enumerate(gold_list):
        absent = [f for f in fields if f not in gold]
        if absent:
            missing_field_rows.append((idx, absent))
        for field in fields:
            if gold.get(field) == [NOT_MENTIONED]:
                placeholder_hits[field] += 1
    total = len(gold_list)
    covered = [f for f, n in placeholder_hits.items() if n > 0]
    return {
        "条数": total,
        "字段不全的条目": missing_field_rows,
        "出现过未提及的字段": covered,
        "未提及覆盖率": len(covered) / len(fields) if fields else 0.0,
    }


def audit_match_set(pairs: list) -> dict:
    """匹配评测集体检。pairs 里每项是 (句子一, 句子二, 金标, 是否难例)。"""
    pos = [p for p in pairs if p[2] == "是"]
    neg = [p for p in pairs if p[2] == "不是"]
    hard_neg = [p for p in neg if p[3]]
    total = len(pairs)
    return {
        "条数": total,
        "正类": len(pos),
        "负类": len(neg),
        "正类占比": len(pos) / total if total else 0.0,
        "难负样本": len(hard_neg),
        "难负占负类": len(hard_neg) / len(neg) if neg else 0.0,
    }


IE_GOLD = [
    {"日期": ["2023-02-15"], "股票名称": ["佰笃[BD]美股"], "开盘价": ["10美元"],
     "收盘价": ["13美元"], "成交量": ["460,000"]},
    {"日期": ["2023-04-05"], "股票名称": ["盘古(0021)"], "开盘价": ["23元"],
     "收盘价": ["26美元"], "成交量": ["310,000"]},
    {"日期": ["2023-05-20"], "股票名称": ["长虹科技"], "开盘价": ["48元"],
     "收盘价": [NOT_MENTIONED], "成交量": ["120,000"]},
    {"日期": ["2023-06-11"], "股票名称": ["华兴能源"], "开盘价": [NOT_MENTIONED],
     "收盘价": ["77元"], "成交量": [NOT_MENTIONED]},
]

MATCH_PAIRS = [
    ("股票市场今日大涨，投资者乐观。", "持续上涨的市场让投资者感到满意。", "是", False),
    ("利率上升，影响房地产市场。", "高利率对房地产有一定冲击。", "是", False),
    ("公司ABC发布季度财报，盈利增长。", "财报披露，公司ABC利润上升。", "是", False),
    ("油价大幅下跌，能源公司面临挑战。", "未来智能城市的建设趋势愈发明显。", "不是", False),
    ("央行降息，刺激经济增长。", "新能源技术的创新。", "不是", False),
    # 难负样本：同一个话题、同一批词，方向相反
    ("央行宣布降息，市场流动性宽松。", "央行宣布加息，市场流动性收紧。", "不是", True),
    # 难负样本：同一家公司，说的是两件事
    ("公司ABC完成对某科技企业的并购。", "公司ABC发布季度财报，盈利增长。", "不是", True),
]


if __name__ == "__main__":
    ie_report = audit_ie_set(IE_GOLD, FIELDS)
    print("【抽取评测集】")
    for k, v in ie_report.items():
        print(f"  {k}: {v}")

    match_report = audit_match_set(MATCH_PAIRS)
    print("\n【匹配评测集】")
    for k, v in match_report.items():
        print(f"  {k}: {v:.3f}" if isinstance(v, float) else f"  {k}: {v}")

    print("\n" + "─" * 46)
    assert ie_report["字段不全的条目"] == [], "金标必须字段齐全，否则字段级打分没法算"
    assert ie_report["未提及覆盖率"] >= 0.5, \
        "至少一半字段要出现过「未提及」，否则测不到抽不到时的行为"
    assert 0.3 <= match_report["正类占比"] <= 0.7, "正负比例过偏，准确率会失真"
    assert match_report["难负占负类"] >= 0.4, \
        "负样本全是风马牛不相及的，分数会虚高"
    print("两份评测集体检通过：它们能测出该测的东西")

# -*- coding: utf-8 -*-
"""信息抽取骨架：换 schema、换例子、换模型接口，其余不用动。

改完 5 处 TODO 就能用。没接真实模型之前，把 call_model 留成假的，
先用 _demo() 把「提示词长什么样、校验拦不拦得住坏数据」验证通过。

纯标准库，`python3 ie_skeleton.py` 直接跑（跑的是离线自检那条路）。
"""
import json
import re

# ── TODO 1：定义 schema —— 要抽哪些字段，一次定死 ───────────────────
SCHEMA = {
    "金融": ["日期", "股票名称", "开盘价", "收盘价", "成交量"],
}
NOT_MENTIONED = "原文中未提及"

# ── TODO 2：提示词骨架 —— 末尾那句格式约束不要删 ─────────────────────
IE_PATTERN = (
    "{}\n\n提取上述句子中{}的实体，并按照JSON格式输出，"
    "上述句子中不存在的信息用['" + NOT_MENTIONED + "']来表示，多个值之间用','分隔。"
)

# ── TODO 3：few-shot 例子 —— 至少一条，答案必须是你想要的那个形状 ─────
EXAMPLES = {
    "金融": [
        {
            "content": "2023-01-10，股市震荡。股票古哥-D[EOOE]美股今日开盘价100美元，"
                       "最终以102美元收盘，成交量达到520000。",
            "answers": {
                "日期": ["2023-01-10"],
                "股票名称": ["古哥-D[EOOE]美股"],
                "开盘价": ["100美元"],
                "收盘价": ["102美元"],
                "成交量": ["520000"],
            },
        }
    ]
}

FENCE_RE = re.compile(r"```(?:json|JSON)?\s*(.*?)```", re.DOTALL)
OBJ_RE = re.compile(r"\{.*\}", re.DOTALL)


def build_prompt(sentence: str, entity_type: str) -> str:
    fields = ", ".join(SCHEMA[entity_type])
    return IE_PATTERN.format(sentence, f"“{entity_type}”({fields})")


def init_prompts(entity_type: str) -> list:
    history = [{"role": "system", "content": "你是一个信息抽取助手。"}]
    for example in EXAMPLES[entity_type]:
        history.append({"role": "user",
                        "content": build_prompt(example["content"], entity_type)})
        history.append({"role": "assistant",
                        "content": json.dumps(example["answers"], ensure_ascii=False)})
    return history


def parse(text: str):
    """去围栏 → 截出对象 → 解析。失败返回 None，绝不返回半成品。"""
    if not isinstance(text, str) or not text.strip():
        return None
    blocks = FENCE_RE.findall(text)
    body = blocks[0].strip() if blocks and blocks[0].strip() else text.strip()
    match = OBJ_RE.search(body)
    if match:
        body = match.group(0)
    try:
        data = json.loads(body)
    except ValueError:
        return None
    return data if isinstance(data, dict) else None


def normalize(data: dict, fields: list) -> dict:
    """字段补齐、值统一成 list[str]、野字段丢掉。只修形状，不猜内容。"""
    fixed = {}
    for field in fields:
        value = data.get(field)
        if value in (None, "", []):
            fixed[field] = [NOT_MENTIONED]
        elif isinstance(value, list):
            items = [str(v) for v in value if str(v).strip()]
            fixed[field] = items or [NOT_MENTIONED]
        else:
            fixed[field] = [str(value)]
    return fixed


def check(data: dict, fields: list) -> list:
    problems = []
    for field in fields:
        value = data.get(field)
        if not isinstance(value, list) or not value:
            problems.append(f"{field} 不是非空列表")
        elif NOT_MENTIONED in value and len(value) > 1:
            problems.append(f"{field} 占位符与真值混用")
    return problems


def call_model(messages: list) -> str:
    """── TODO 4：接上真实模型 ───────────────────────────────────

    本地 Ollama 的写法（需要 pip install ollama，并先 ollama run <模型>）：

        import ollama, os
        model = os.environ.get("OLLAMA_MODEL", "qwen3:8b")
        return ollama.chat(model=model, messages=messages)["message"]["content"]

    换成别的服务时，只需保证这个函数「收消息列表、返回一段字符串」。
    密钥一律走 os.environ.get(...)，不要写进源码。
    """
    raise NotImplementedError("接上模型后删掉这一行")


def extract(sentence: str, entity_type: str, retries: int = 2) -> dict:
    history = init_prompts(entity_type)
    fields = SCHEMA[entity_type]
    raw = ""
    for attempt in range(retries + 1):
        messages = list(history) + [
            {"role": "user", "content": build_prompt(sentence, entity_type)}
        ]
        if attempt > 0:
            messages.append({"role": "assistant", "content": raw})
            messages.append({"role": "user",
                             "content": "上一条不是合法 JSON，请只输出一个 JSON 对象。"})
        raw = call_model(messages)
        data = parse(raw)
        if data is None:
            continue
        data = normalize(data, fields)
        if not check(data, fields):
            return data
    raise ValueError(f"重试 {retries} 次仍未拿到合法结果：{raw[:120]}")


def _demo():
    """── TODO 5：接上真实模型后，把这里换成一条真实句子的端到端冒烟测试 ──"""
    fields = SCHEMA["金融"]
    print("提示词长这样：\n")
    print(build_prompt("2023-02-15，股票佰笃[BD]美股开盘价10美元。", "金融"))

    fake = '```json\n{"日期": ["2023-02-15"], "股票名称": "佰笃[BD]美股", "开盘价": ["10美元"]}\n```'
    data = parse(fake)
    assert data is not None, "带围栏的输出应当能解析"
    fixed = normalize(data, fields)
    assert check(fixed, fields) == []
    assert fixed["收盘价"] == [NOT_MENTIONED], "缺的字段要补占位符，不能漏键"
    print("\n规整后：", json.dumps(fixed, ensure_ascii=False))
    assert parse("我不知道") is None, "解析不了必须是 None"
    print("离线自检通过：提示词、解析、规整、校验四段都在")


if __name__ == "__main__":
    _demo()

# -*- coding: utf-8 -*-
"""文本匹配骨架：换判据、换例子、换模型接口，其余不用动。

改完 5 处 TODO 就能用。没接真实模型之前，_demo() 会把
「回答归一化」和「阈值选择」两段离线验证通过。

纯标准库，`python3 match_skeleton.py` 直接跑（跑的是离线自检那条路）。
"""
POSITIVE = "是"
NEGATIVE = "不是"

# ── TODO 1：写清楚「同一件事」的边界 —— 这句话决定了全部标注口径 ──────
SYSTEM_PROMPT = (
    "现在你需要帮助我完成文本匹配任务，当我给你两个句子时，"
    "你需要回答我这两句话语义是否相似。只需要回答是否相似，不要做多余的回答。"
)

# ── TODO 2：few-shot —— 正负都要给，负样本里要有「像但不是」的难例 ────
EXAMPLES = {
    POSITIVE: [
        ("公司ABC发布了季度财报，显示盈利增长。", "财报披露，公司ABC利润上升。"),
    ],
    NEGATIVE: [
        ("黄金价格下跌，投资者抛售。", "外汇市场交易额创下新高。"),
        ("央行宣布降息，市场流动性宽松。", "央行宣布加息，市场流动性收紧。"),
    ],
}

PAIR_PATTERN = "句子一: {}\n句子二: {}\n上面两句话是相似的语义吗？"


def init_prompts() -> list:
    history = [{"role": "system", "content": SYSTEM_PROMPT}]
    for label, pairs in EXAMPLES.items():
        for s1, s2 in pairs:
            history.append({"role": "user", "content": PAIR_PATTERN.format(s1, s2)})
            history.append({"role": "assistant", "content": label})
    return history


def normalize_answer(text: str):
    """── TODO 3：把模型的自由回答收敛成两个值之一 ───────────────────

    先判否定词，再判肯定词。顺序反了，「不是」会被「是」吃掉。
    收敛不了就返回 None —— 判不了要显式暴露，不许默认归到某一类。
    """
    if not isinstance(text, str):
        return None
    cleaned = text.strip().replace(" ", "").replace("\n", "")
    for marker in ("不是", "不相似", "不同", "否"):
        if marker in cleaned:
            return NEGATIVE
    for marker in ("是", "相似", "相同"):
        if marker in cleaned:
            return POSITIVE
    return None


def call_model(messages: list) -> str:
    """── TODO 4：接上真实模型 ───────────────────────────────────

    本地 Ollama 的写法（需要 pip install ollama，并先 ollama run <模型>）：

        import ollama, os
        model = os.environ.get("OLLAMA_MODEL", "qwen3:8b")
        return ollama.chat(model=model, messages=messages)["message"]["content"]

    密钥一律走 os.environ.get(...)，不要写进源码。
    """
    raise NotImplementedError("接上模型后删掉这一行")


def match_pair(s1: str, s2: str, history: list, retries: int = 1):
    raw = ""
    for attempt in range(retries + 1):
        messages = list(history) + [{"role": "user", "content": PAIR_PATTERN.format(s1, s2)}]
        if attempt > 0:
            messages.append({"role": "user",
                             "content": f"请只回答「{POSITIVE}」或「{NEGATIVE}」。"})
        raw = call_model(messages)
        answer = normalize_answer(raw)
        if answer is not None:
            return answer, raw
    return None, raw


def evaluate(gold: list, pred: list) -> dict:
    """精确率 / 召回率 / F1。准确率单独看会被样本比例带偏，所以不单独报。"""
    tp = sum(1 for g, p in zip(gold, pred) if g == POSITIVE and p == POSITIVE)
    fp = sum(1 for g, p in zip(gold, pred) if g == NEGATIVE and p == POSITIVE)
    fn = sum(1 for g, p in zip(gold, pred) if g == POSITIVE and p == NEGATIVE)
    precision = tp / (tp + fp) if (tp + fp) else 0.0
    recall = tp / (tp + fn) if (tp + fn) else 0.0
    f1 = 2 * precision * recall / (precision + recall) if (precision + recall) else 0.0
    return {"精确率": precision, "召回率": recall, "F1": f1}


def pick_threshold(scored: list, candidates: list) -> float:
    """── TODO 5：拿到分数时用这个挑线；只有二选一回答时删掉本函数 ─────

    scored 里每项是 (金标, 分数)。挑 F1 最高的阈值，并列时取更高的那个
    （更保守，宁可放过不肯错抓）。
    """
    gold = [label for label, _ in scored]
    best_th, best_f1 = candidates[0], -1.0
    for th in candidates:
        pred = [POSITIVE if score >= th else NEGATIVE for _, score in scored]
        f1 = evaluate(gold, pred)["F1"]
        if f1 >= best_f1:
            best_th, best_f1 = th, f1
    return best_th


def _demo():
    print("前置消息共", len(init_prompts()), "条")

    # 归一化：否定词优先，这是最容易写反的一处
    assert normalize_answer("不是") == NEGATIVE
    assert normalize_answer("是的，两句话语义相似") == POSITIVE
    assert normalize_answer("这两句话不相似。") == NEGATIVE
    assert normalize_answer("嗯……不好说") is None, "判不了要返回 None"
    print("回答归一化自检通过（含「不是」不会被「是」吃掉）")

    scored = [("是", 0.9), ("是", 0.7), ("不是", 0.6), ("不是", 0.2), ("是", 0.65)]
    th = pick_threshold(scored, [round(0.1 * i, 1) for i in range(1, 10)])
    pred = [POSITIVE if s >= th else NEGATIVE for _, s in scored]
    metric = evaluate([g for g, _ in scored], pred)
    print(f"选出的阈值 {th}，此时 {metric}")
    assert metric["F1"] >= 0.8
    print("阈值选择自检通过：阈值来自评测集，不来自手感")


if __name__ == "__main__":
    _demo()