【案例】金融行业动态方向评估（一）：方法论与文本分类

# -*- coding: utf-8 -*-
"""把模型吐出来的自由文本收敛成一个枚举值，越界的显式报出来。

大模型返回的是字符串，不是枚举。它可能回「财务报告」，也可能回
「这段文字属于财务报告类别，因为……」，还可能回一个你压根没定义过的
「财报」。程序如果直接拿这串东西当标签，准确率会被格式噪声吃掉一大截。

这份文件只做一件事：**文本 → 枚举值 或者 明确的「判不了」**。
纯标准库可跑。
"""
import re

LABELS = ["新闻报道", "财务报告", "公司公告", "分析师报告"]

# 常见的同义/简写写法，收敛到正式类别名。
# 注意：这张表是「容错」不是「猜」——只收录含义唯一的写法。
ALIAS = {
    "财报": "财务报告",
    "财务公告": "财务报告",
    "年报": "财务报告",
    "季报": "财务报告",
    "新闻": "新闻报道",
    "报道": "新闻报道",
    "公告": "公司公告",
    "研报": "分析师报告",
    "分析报告": "分析师报告",
    "研究报告": "分析师报告",
}

# 模型爱加的包装：思考块、代码围栏、引号、句末标点
_THINK = re.compile(r"<think>.*?</think>", re.S)
_FENCE = re.compile(r"```[a-zA-Z]*\s*(.*?)\s*```", re.S)
_TRIM = "\"'“”‘’ 。．.，,：:；;！!？?\n\t "


def normalize(raw):
    """清掉包装，返回一串干净文本。不做任何类别判断。"""
    if raw is None:
        return ""
    text = str(raw)
    text = _THINK.sub("", text)          # 带 thinking 能力的模型会先吐一段推理
    m = _FENCE.search(text)
    if m:
        text = m.group(1)                # 代码围栏里才是正文
    return text.strip(_TRIM)


def parse_label(raw, labels=None, allow_alias=True):
    """返回 (类别 或 None, 判定依据)。

    判定顺序是有讲究的，从严到松：
      1. exact  —— 清洗后正好等于某个类别名，这是唯一「干净」的情况；
      2. alias  —— 命中同义表；
      3. unique —— 文本里只出现了一个类别名（模型加了解释但没改口）；
      4. 其余一律返回 None，绝不猜。
    """
    labels = list(labels or LABELS)
    text = normalize(raw)
    if not text:
        return None, "empty"

    if text in labels:
        return text, "exact"

    if allow_alias and text in ALIAS and ALIAS[text] in labels:
        return ALIAS[text], "alias"

    # 模型没忍住加了解释：只有当文本里**恰好只出现一个**类别名时才接受
    hit = [lab for lab in labels if lab in text]
    if len(hit) == 1:
        return hit[0], "unique"
    if len(hit) > 1:
        # 出现多个类别名，说明模型在比较或者在犹豫，这种绝不能挑一个用
        return None, "ambiguous"

    if allow_alias:
        ahit = {ALIAS[k] for k in ALIAS if k in text and ALIAS[k] in labels}
        if len(ahit) == 1:
            return ahit.pop(), "alias_in_text"

    return None, "out_of_scope"


def _demo():
    cases = [
        ("财务报告", "干净的枚举值"),
        ("  财务报告。 ", "带空白和句号"),
        ("“财务报告”", "带引号"),
        ("```\n财务报告\n```", "被代码围栏包住"),
        ("<think>先看有没有资产负债表</think>财务报告", "带思考块"),
        ("财报", "同义写法"),
        ("财务公告", "选项和示例用词不一致时的典型产物"),
        ("这段文字属于财务报告类别，因为它描述了资产负债表", "多了解释，但只提到一个类别"),
        ("可能是财务报告，也可能是公司公告", "两个类别都提到了"),
        ("董事会决议", "压根不在类别体系里"),
        ("", "空返回"),
    ]
    print("%-46s %-12s %s" % ("模型返回（截断显示）", "解析结果", "依据"))
    print("-" * 78)
    ok = 0
    for raw, _note in cases:
        lab, why = parse_label(raw)
        shown = (raw[:20] + "…") if len(raw) > 21 else raw
        print("%-46s %-12s %s" % (repr(shown), lab if lab else "判不了", why))
        ok += int(lab is not None)
    print("-" * 78)
    print("%d/%d 条能收敛到枚举值，其余显式判不了。" % (ok, len(cases)))

    # 断言：这几条是安全底线，改代码把它们改坏了要当场炸
    assert parse_label("财务报告")[0] == "财务报告"
    assert parse_label("可能是财务报告，也可能是公司公告") == (None, "ambiguous")
    assert parse_label("董事会决议") == (None, "out_of_scope")
    assert parse_label("")[0] is None
    assert parse_label("财报", allow_alias=False)[0] is None, \
        "关掉同义表后就不该再认简写"
    print("越界处理自检通过：模糊与越界都返回 None，没有一条被硬塞进某一类。")


if __name__ == "__main__":
    _demo()

# -*- coding: utf-8 -*-
"""金融文本方向判定的评测集：24 条带标准答案的样本。

这份文件是整条链路上唯一「先于提示词存在」的东西。改提示词之前它必须
已经躺在仓库里，否则后面所有的准确率都没有参照物。

设计约定：
  · 四个类别每类 6 条，人为保证类别均衡，算总准确率时不会被大类带偏；
  · 每条都带 sid，出了问题能一眼定位到是哪一条；
  · hard=True 标记边界样本（两个类别都沾边的那种），单独统计；
  · 纯标准库，import 进来就能用，不依赖任何模型。
"""

# 类别体系：全流程只认这四个字符串，多一个字少一个字都算越界
LABELS = ["新闻报道", "财务报告", "公司公告", "分析师报告"]

# 每条样本：(sid, 文本, 标准答案, 是否边界样本)
SAMPLES = [
    # ---------- 新闻报道：第三方视角在陈述已经发生的事 ----------
    ("N01", "今日，央行发布公告宣布降低利率，以刺激经济增长。这一降息举措将影响贷款利率，并在未来几个季度内对金融市场产生影响。", "新闻报道", False),
    ("N02", "今日，股市经历了一轮震荡，受到宏观经济数据和全球贸易紧张局势的影响。投资者密切关注美联储可能的政策调整，以适应市场的不确定性。", "新闻报道", False),
    ("N03", "受隔夜外围市场影响，今日开盘三大股指集体低开，随后震荡回升，尾盘收复大部分失地。", "新闻报道", False),
    ("N04", "据多家媒体报道，某大型券商昨日下午召开临时会议，讨论调整旗下多只产品的费率结构。", "新闻报道", False),
    ("N05", "监管部门今日就近期市场波动答记者问，表示将继续维护市场平稳运行。", "新闻报道", True),
    ("N06", "本周大宗商品市场普遍回落，铜、铝价格分别下跌，相关板块个股跟随走弱。", "新闻报道", False),

    # ---------- 财务报告：企业自己在披露经营与财务数据 ----------
    ("F01", "本公司年度财务报告显示，去年公司实现了稳步增长的盈利，同时资产负债表呈现强劲的状况。经济环境的稳定和管理层的有效战略执行为公司的健康发展奠定了基础。", "财务报告", False),
    ("F02", "公司资产负债表显示，公司偿债能力强劲，现金流充足，为未来投资和扩张提供了坚实的财务基础。", "财务报告", False),
    ("F03", "报告期内，公司实现营业收入同比增长，归属于母公司股东的净利润同步提升，毛利率较上年同期略有改善。", "财务报告", False),
    ("F04", "本季度经营活动产生的现金流量净额为正，存货周转天数较上季度缩短，应收账款账龄结构保持稳定。", "财务报告", False),
    ("F05", "截至报告期末，公司资产总额较年初增长，负债率维持在行业平均水平以下。", "财务报告", False),
    ("F06", "公司管理层在年报中对下一年度的经营目标作出说明，并披露了主要财务指标的完成情况。", "财务报告", True),

    # ---------- 公司公告：企业在宣布一件具体的事项 ----------
    ("A01", "本公司高兴地宣布成功完成最新一轮并购交易，收购了一家在人工智能领域领先的公司。这一战略举措将有助于扩大我们的业务领域，提高市场竞争力", "公司公告", False),
    ("A02", "本公司宣布成功收购一家在创新科技领域领先的公司，这一战略性收购将有助于公司拓展技术能力和加速产品研发。", "公司公告", False),
    ("A03", "本公司董事会决议通过，聘任新任首席财务官，自决议之日起生效，相关手续正在办理中。", "公司公告", False),
    ("A04", "本公司拟于下月召开临时股东大会，审议关于变更募集资金用途的议案，现将会议相关事项通知如下。", "公司公告", False),
    ("A05", "本公司控股子公司近日取得一项发明专利授权，该专利的取得不会对公司当期经营业绩产生重大影响。", "公司公告", False),
    ("A06", "本公司发布关于回购股份进展的说明，截至目前已累计回购股份若干股。", "公司公告", True),

    # ---------- 分析师报告：第三方在给判断与建议 ----------
    ("R01", "最新的行业分析报告指出，科技公司的创新将成为未来增长的主要推动力。云计算、人工智能和数字化转型被认为是引领行业发展的关键因素，投资者应关注这些趋势", "分析师报告", False),
    ("R02", "最新的分析报告指出，可再生能源行业预计将在未来几年经历持续增长，投资者应该关注这一领域的投资机会", "分析师报告", False),
    ("R03", "我们维持对该板块的中性评级，认为当前估值已部分反映了行业景气度的改善预期。", "分析师报告", False),
    ("R04", "研究团队认为，随着渗透率提升放缓，该细分赛道的竞争格局将在未来两年内逐步明朗。", "分析师报告", False),
    ("R05", "本报告基于公开资料整理，测算结果显示该行业未来三年复合增速有望维持在两位数区间。", "分析师报告", False),
    ("R06", "综合公司披露的经营数据与行业景气度，我们上调了对其盈利能力的判断。", "分析师报告", True),
]


def labels():
    """返回类别体系的副本，避免调用方误改全局常量。"""
    return list(LABELS)


def samples(only_hard=None):
    """取样本。only_hard=True 只取边界样本，False 只取普通样本，None 全取。"""
    if only_hard is None:
        return list(SAMPLES)
    return [s for s in SAMPLES if s[3] is bool(only_hard)]


def self_check():
    """评测集自身的体检。数据有毛病要在跑模型之前就炸出来。"""
    sids = [s[0] for s in SAMPLES]
    assert len(sids) == len(set(sids)), "sid 重复，定位样本时会串"

    texts = [s[1] for s in SAMPLES]
    assert len(texts) == len(set(texts)), "有完全重复的文本，会让准确率虚高"

    for sid, text, gold, hard in SAMPLES:
        assert gold in LABELS, "%s 的标准答案 %r 不在类别体系里" % (sid, gold)
        assert text.strip(), "%s 的文本是空的" % sid
        assert isinstance(hard, bool), "%s 的 hard 标记不是布尔值" % sid

    # 类别必须均衡，否则「总准确率」这个数会被样本多的类别绑架
    per = {}
    for _sid, _t, gold, _h in SAMPLES:
        per[gold] = per.get(gold, 0) + 1
    assert len(set(per.values())) == 1, "各类别样本数不相等：%s" % per
    return per


if __name__ == "__main__":
    per = self_check()
    print("评测集自检通过")
    print("  样本总数 : %d" % len(SAMPLES))
    print("  类别数   : %d" % len(LABELS))
    for lab in LABELS:
        print("    %-6s %d 条" % (lab, per[lab]))
    print("  边界样本 : %d 条（%s）"
          % (len(samples(True)), " ".join(s[0] for s in samples(True))))

# -*- coding: utf-8 -*-
"""提示词渲染器：把「角色 + 任务 + 类别 + 格式 + 示例」拼成 messages。

把提示词当字符串手写，改三次就会乱；把它当**数据结构渲染出来**，
改哪一处、改了几处，都能被程序检查。这份文件不调任何模型，
它只负责把配置渲染成 messages 列表并打印出来。纯标准库可跑。
"""
import json

import evalset

# 每个类别一条示例。示例文本直接取自评测集之外的那几条典型样本，
# 目的是让模型看到「什么样的文字算这一类」，而不是靠类别名猜。
CLASS_EXAMPLES = {
    "新闻报道": "今日，股市经历了一轮震荡，受到宏观经济数据和全球贸易紧张局势的影响。投资者密切关注美联储可能的政策调整，以适应市场的不确定性。",
    "财务报告": "本公司年度财务报告显示，去年公司实现了稳步增长的盈利，同时资产负债表呈现强劲的状况。经济环境的稳定和管理层的有效战略执行为公司的健康发展奠定了基础。",
    "公司公告": "本公司高兴地宣布成功完成最新一轮并购交易，收购了一家在人工智能领域领先的公司。这一战略举措将有助于扩大我们的业务领域，提高市场竞争力",
    "分析师报告": "最新的行业分析报告指出，科技公司的创新将成为未来增长的主要推动力。云计算、人工智能和数字化转型被认为是引领行业发展的关键因素，投资者应关注这些趋势",
}

ASK = "“{text}”是 {labels} 里的什么类别？"


def build_system(labels, with_format_rule=True, with_boundary=False):
    """角色 + 任务 + 类别体系（+ 输出格式约束）（+ 边界说明）。"""
    parts = ["现在你是一个文本分类器，你需要按照要求将我给你的句子分类到：%s类别中。" % labels]
    if with_format_rule:
        parts.append("只输出类别名本身，不要输出任何解释、理由或标点。")
    if with_boundary:
        parts.append(
            "判定依据是这段文字的**体裁**，不是它提到的内容："
            "第三方在陈述已经发生的事是新闻报道；"
            "企业自己在披露经营与财务数据是财务报告；"
            "企业自己在宣布一件具体事项是公司公告；"
            "第三方在给出判断与预期是分析师报告。")
    return "".join(parts)


def build_messages(text, labels, shots=None,
                   with_format_rule=True, with_boundary=False):
    """渲染出完整 messages。shots 为 None 时就是 zero-shot。"""
    msgs = [{"role": "system",
             "content": build_system(labels, with_format_rule, with_boundary)}]
    for label, example in (shots or {}).items():
        msgs.append({"role": "user", "content": ASK.format(text=example, labels=labels)})
        msgs.append({"role": "assistant", "content": label})
    msgs.append({"role": "user", "content": ASK.format(text=text, labels=labels)})
    return msgs


def describe(msgs):
    """粗略统计规模。字符数不是 token 数，但足够看出量级变化。"""
    chars = sum(len(m["content"]) for m in msgs)
    roles = {}
    for m in msgs:
        roles[m["role"]] = roles.get(m["role"], 0) + 1
    return {"messages": len(msgs), "chars": chars, "roles": roles}


def _demo():
    labels = evalset.labels()
    target = "公司资产负债表显示，公司偿债能力强劲，现金流充足，为未来投资和扩张提供了坚实的财务基础。"

    variants = [
        ("v1 只有角色和类别", dict(shots=None, with_format_rule=False, with_boundary=False)),
        ("v2 加一句格式约束", dict(shots=None, with_format_rule=True, with_boundary=False)),
        ("v3 再加四条示例", dict(shots=CLASS_EXAMPLES, with_format_rule=True, with_boundary=False)),
        ("v4 再补边界说明", dict(shots=CLASS_EXAMPLES, with_format_rule=True, with_boundary=True)),
    ]

    print("【一、四版提示词的规模】")
    print("  %-18s %8s %8s  %s" % ("版本", "消息条数", "字符数", "角色分布"))
    prev = None
    for name, kw in variants:
        msgs = build_messages(target, labels, **kw)
        d = describe(msgs)
        delta = "" if prev is None else "  (+%d 字符)" % (d["chars"] - prev)
        print("  %-18s %8d %8d  %s%s"
              % (name, d["messages"], d["chars"],
                 " ".join("%s×%d" % (k, v) for k, v in sorted(d["roles"].items())),
                 delta))
        prev = d["chars"]

    print("\n【二、v3 渲染出来长什么样】")
    msgs = build_messages(target, labels, shots=CLASS_EXAMPLES)
    for i, m in enumerate(msgs):
        body = m["content"]
        body = body if len(body) <= 52 else body[:52] + "…"
        print("  [%02d] %-9s %s" % (i, m["role"], body))

    print("\n【三、示例对的方向不能反】")
    pairs = [(msgs[i], msgs[i + 1]) for i in range(1, len(msgs) - 1, 2)]
    for u, a in pairs:
        assert u["role"] == "user" and a["role"] == "assistant", \
            "示例必须是 user 问、assistant 答，反了模型就学成了反向映射"
        assert a["content"] in labels, "示例答案 %r 不在类别体系里" % a["content"]
    print("  %d 组示例对方向正确，答案全部落在类别体系内。" % len(pairs))

    # 断言：v1→v2 只能差那一句格式约束，别的一个字都不许动
    m1 = build_messages(target, labels, shots=None, with_format_rule=False)
    m2 = build_messages(target, labels, shots=None, with_format_rule=True)
    assert len(m1) == len(m2), "v1 与 v2 的消息条数必须一致"
    assert m1[-1] == m2[-1], "待判句这一条不许跟着变"
    added = m2[0]["content"].replace(m1[0]["content"], "")
    assert added == "只输出类别名本身，不要输出任何解释、理由或标点。", \
        "v1→v2 改动不止一处：%r" % added
    print("\n  v1→v2 的全部差异：%r" % added)
    print("  就这一句，别的一个字没动——这才叫「一次只改一处」。")

    print("\n【四、messages 原样（v2，可直接贴进请求体）】")
    print(json.dumps(m2, ensure_ascii=False, indent=2)[:600] + "\n  …")


if __name__ == "__main__":
    _demo()

# -*- coding: utf-8 -*-
"""评测集打分器：准确率、每类 P/R/F1、混淆矩阵、误差区间。

输入是「标准答案 + 预测」两列，跟模型、跟接口、跟哪一版提示词都无关，
所以它可以在没有任何推理环境的机器上跑。纯标准库。

用法：
    python3 eval_scorer.py              # 用内置的一组演示预测跑一遍
    from eval_scorer import score       # 当成模块用
"""
import math

import evalset


def score(pairs, labels):
    """pairs: [(sid, 标准答案, 预测 或 None), ...]

    预测为 None 表示模型没给出合法枚举值（越界/模糊/空），
    它**算错，但单独计数**——这两种错的修法完全不同。
    """
    total = len(pairs)
    right = sum(1 for _s, g, p in pairs if p is not None and p == g)
    invalid = sum(1 for _s, _g, p in pairs if p is None)

    # 混淆矩阵：cm[真实][预测]，预测列多一个 "判不了"
    cols = list(labels) + ["判不了"]
    cm = {g: {c: 0 for c in cols} for g in labels}
    for _sid, gold, pred in pairs:
        cm[gold][pred if pred is not None else "判不了"] += 1

    per = {}
    for lab in labels:
        tp = cm[lab][lab]
        fn = sum(cm[lab][c] for c in cols) - tp
        fp = sum(cm[g][lab] for g in labels if g != lab)
        prec = tp / float(tp + fp) if (tp + fp) else 0.0
        rec = tp / float(tp + fn) if (tp + fn) else 0.0
        f1 = 2 * prec * rec / (prec + rec) if (prec + rec) else 0.0
        per[lab] = {"tp": tp, "fp": fp, "fn": fn,
                    "precision": prec, "recall": rec, "f1": f1}

    acc = right / float(total) if total else 0.0
    return {"total": total, "right": right, "invalid": invalid,
            "accuracy": acc, "per_label": per, "cm": cm, "cols": cols}


def ci95(acc, n):
    """准确率的 95% 置信区间半宽（正态近似）。

    用途只有一个：判断两版提示词的差距是不是噪声。
    n 很小的时候这个半宽会大得吓人——那说明评测集该扩了，不是提示词不行。
    """
    if n <= 0:
        return 0.0
    return 1.96 * math.sqrt(acc * (1 - acc) / float(n))


def report(res, labels, title="评测结果"):
    lines = []
    half = ci95(res["accuracy"], res["total"])
    lines.append("【%s】" % title)
    lines.append("  样本数     : %d" % res["total"])
    lines.append("  判对       : %d" % res["right"])
    lines.append("  判不了     : %d（越界/模糊/空返回，计入错误）" % res["invalid"])
    lines.append("  准确率     : %.1f%%   95%% 区间约 ±%.1f 个百分点"
                 % (res["accuracy"] * 100, half * 100))
    lines.append("")
    lines.append("  %-8s %8s %8s %8s   (TP/FP/FN)" % ("类别", "精确率", "召回率", "F1"))
    for lab in labels:
        d = res["per_label"][lab]
        lines.append("  %-8s %7.1f%% %7.1f%% %7.1f%%   (%d/%d/%d)"
                     % (lab, d["precision"] * 100, d["recall"] * 100,
                        d["f1"] * 100, d["tp"], d["fp"], d["fn"]))
    lines.append("")
    lines.append("  混淆矩阵（行=标准答案，列=预测）")
    head = "  %-10s" % "" + "".join("%-9s" % c for c in res["cols"])
    lines.append(head)
    for g in labels:
        row = "  %-10s" % g + "".join("%-9d" % res["cm"][g][c] for c in res["cols"])
        lines.append(row)
    return "\n".join(lines)


# 一组演示预测：把「加了解释导致取不出标签」和「相邻类别混淆」两种典型
# 错误都放进来，好让混淆矩阵真的有东西可看。
DEMO_PRED = {
    "N05": "公司公告",      # 监管答记者问被当成了公告
    "F06": "分析师报告",    # 年报里的目标说明被当成了判断
    "A06": None,            # 越界：模型回了个类别体系外的说法
    "R06": "财务报告",      # 引用了经营数据就被拉去财务报告
    "N03": None,            # 模糊：模型同时提到两个类别
}


def _demo():
    labels = evalset.labels()
    evalset.self_check()

    pairs = []
    for sid, _text, gold, _hard in evalset.samples():
        pred = DEMO_PRED[sid] if sid in DEMO_PRED else gold
        pairs.append((sid, gold, pred))

    res = score(pairs, labels)
    print(report(res, labels, "演示预测"))

    print("")
    print("  边界样本单独看：")
    hard_sids = {s[0] for s in evalset.samples(True)}
    hard_pairs = [p for p in pairs if p[0] in hard_sids]
    hres = score(hard_pairs, labels)
    print("    边界 %d 条，判对 %d 条，准确率 %.1f%%"
          % (hres["total"], hres["right"], hres["accuracy"] * 100))
    easy_pairs = [p for p in pairs if p[0] not in hard_sids]
    eres = score(easy_pairs, labels)
    print("    普通 %d 条，判对 %d 条，准确率 %.1f%%"
          % (eres["total"], eres["right"], eres["accuracy"] * 100))
    print("    总准确率被边界样本拉低了 %.1f 个百分点。"
          % ((eres["accuracy"] - res["accuracy"]) * 100))

    # 断言：打分器自己不能算错
    assert res["total"] == 24
    assert res["right"] + sum(1 for _s, g, p in pairs if p != g) == 24
    assert res["invalid"] == 2, "两条 None 必须被单独数出来"
    for lab in labels:
        s = sum(res["cm"][lab][c] for c in res["cols"])
        assert s == 6, "%s 行的样本数应为 6，实得 %d" % (lab, s)
    perfect = score([(s, g, g) for s, g, _p in pairs], labels)
    assert abs(perfect["accuracy"] - 1.0) < 1e-9, "全对时准确率必须是 100%"
    assert abs(ci95(1.0, 24)) < 1e-9, "准确率 100% 时正态近似区间退化为 0"
    print("\n  打分器自检通过。")


if __name__ == "__main__":
    _demo()

# -*- coding: utf-8 -*-
"""迭代记录器：逼着自己「一次只改一处」，并把效果变化算成数字。

调提示词最常见的失败不是改错了，是**一次改了三处**：加了格式约束、
换了措辞、又补了两个示例，然后准确率涨了 8 个点，你不知道是哪一处起的作用，
下一轮也就无从下手。

这份文件做两件事：
  1. 记录每一版改了哪些字段，改动超过一处直接抛错；
  2. 把相邻两版的指标差算出来，并用置信区间判断这个差是不是噪声。
纯标准库可跑。
"""
from eval_scorer import ci95

# 提示词被拆成可比较的字段。改提示词 = 改这张表里的某一个键。
BASE = {
    "role": "文本分类器",
    "labels": "['新闻报道', '财务报告', '公司公告', '分析师报告']",
    "format_rule": "",
    "shots": 0,
    "boundary_rule": "",
}


class Iteration(object):
    def __init__(self, name, fields, note=""):
        self.name = name
        self.fields = dict(fields)
        self.note = note
        self.metrics = {}

    def diff(self, other):
        """返回两版之间所有不同的字段名。"""
        keys = set(self.fields) | set(other.fields)
        return sorted(k for k in keys
                      if self.fields.get(k) != other.fields.get(k))


def chain(*iterations):
    """检查整条迭代链：相邻两版之间**只准差一个字段**。"""
    for prev, cur in zip(iterations, iterations[1:]):
        changed = prev.diff(cur)
        if len(changed) != 1:
            raise AssertionError(
                "%s → %s 一次改了 %d 处（%s），拆开重做"
                % (prev.name, cur.name, len(changed), "、".join(changed)))
    return True


def compare(prev, cur, n):
    """相邻两版的指标差，附一句「这个差算不算数」的判断。"""
    out = []
    for key in ("parse_rate", "accuracy"):
        a = prev.metrics.get(key)
        b = cur.metrics.get(key)
        if a is None or b is None:
            continue
        delta = b - a
        # 两个比例之差的粗略噪声尺度：各自区间半宽的和
        noise = ci95(a, n) + ci95(b, n)
        verdict = "超出噪声，算数" if abs(delta) > noise else "落在噪声里，不算数"
        out.append((key, a, b, delta, noise, verdict))
    return out


def _demo():
    n = 24  # 评测集大小

    v1 = Iteration("v1", BASE, "只有角色和类别体系")
    v2 = Iteration("v2", dict(BASE, format_rule="只输出类别名本身，不要输出任何解释、理由或标点。"),
                   "加一句输出格式约束")
    v3 = Iteration("v3", dict(v2.fields, shots=4), "每个类别补一条示例")
    v4 = Iteration("v4", dict(v3.fields, boundary_rule="按体裁判定，不按提到的内容判定"),
                   "补类别边界说明")

    print("【一、改动链路检查】")
    chain(v1, v2, v3, v4)
    for prev, cur in ((v1, v2), (v2, v3), (v3, v4)):
        print("  %s → %s  改动字段：%s   %s"
              % (prev.name, cur.name, "、".join(prev.diff(cur)), cur.note))
    print("  四版全部满足「一次只改一处」。")

    print("\n【二、一次改两处会怎样】")
    bad = Iteration("v3x", dict(v2.fields, shots=4,
                                boundary_rule="按体裁判定"), "同时加示例和边界说明")
    try:
        chain(v2, bad)
    except AssertionError as e:
        print("  被拦下：%s" % e)

    print("\n【三、把指标填进去再比（以下为举例用的假想值）】")
    # 下面这组数字是**举例用的假想值**，只用来演示「差多少才算数」
    # 这个判据，它不是任何模型的成绩。真实数字要由 run_eval.py 在你自己的
    # 推理环境里跑出来再回填；在那之前，不要拿这里的值对外报指标。
    v1.metrics = {"parse_rate": 0.50, "accuracy": 0.50}
    v2.metrics = {"parse_rate": 1.00, "accuracy": 0.75}
    v3.metrics = {"parse_rate": 1.00, "accuracy": 0.79}

    for prev, cur in ((v1, v2), (v2, v3)):
        print("  %s → %s" % (prev.name, cur.name))
        for key, a, b, delta, noise, verdict in compare(prev, cur, n):
            print("    %-10s %.0f%% → %.0f%%   差 %+.0f 个百分点，"
                  "噪声尺度 ±%.0f —— %s"
                  % (key, a * 100, b * 100, delta * 100, noise * 100, verdict))

    print("\n【四、这条链路给出的结论】")
    print("  · 格式约束这一处，把「取不出标签」整类问题一次清掉；")
    print("  · 示例这一处，带来的准确率变化落在噪声里，n=24 的评测集判不出来；")
    print("  · 想判出几个点的差别，评测集得先扩大——不是再去改提示词。")

    # 断言：评测集越小，噪声尺度越大，这是扩评测集的硬理由
    assert ci95(0.8, 24) > ci95(0.8, 240), "样本越多，区间应越窄"
    assert ci95(0.8, 24) > 0.15, "n=24 时半宽超过 15 个百分点"
    print("\n  n=24 时半宽 ±%.1f 个百分点；n=240 时收窄到 ±%.1f 个百分点。"
          % (ci95(0.8, 24) * 100, ci95(0.8, 240) * 100))


if __name__ == "__main__":
    _demo()

# -*- coding: utf-8 -*-
"""金融文本方向判定：可以上生产的那一版。

跟随手写的脚本比，这一版多了四件事，每一件都是线上真会出问题的地方：
  1. 模型名、地址、超时、重试全部走环境变量，不写死在代码里；
  2. 每一次调用都包了异常处理，单条失败不拖垮整批；
  3. 模型返回先过 output_guard，收不进枚举值就记成「判不了」，不硬塞；
  4. 结果带上原始返回，出了问题能回看模型到底说了什么。

运行前置：
    pip install ollama rich
    ollama run qwen3:8b          # 先把模型拉到本地
可选环境变量：
    FIN_MODEL       模型名，默认 qwen3:8b
    OLLAMA_HOST     Ollama 服务地址，默认 http://127.0.0.1:11434
    FIN_TIMEOUT     单次调用超时秒数，默认 60
    FIN_RETRY       单条最多重试次数，默认 2
"""
import os
import sys
import time

from rich import print
from rich.console import Console

import evalset
from output_guard import parse_label
from prompt_render import CLASS_EXAMPLES, build_messages

console = Console()

MODEL = os.environ.get("FIN_MODEL", "qwen3:8b")
HOST = os.environ.get("OLLAMA_HOST", "http://127.0.0.1:11434")
TIMEOUT = float(os.environ.get("FIN_TIMEOUT", "60"))
RETRY = int(os.environ.get("FIN_RETRY", "2"))


def _client():
    """延迟导入 ollama：没装依赖时给一句人话，而不是一屏 traceback。"""
    try:
        import ollama
    except ImportError:
        print("[bold red]没有找到 ollama 库，先执行 pip install ollama[/bold red]")
        sys.exit(1)
    # 指定 host 才能连远端的推理机；不指定就是本机默认端口
    return ollama.Client(host=HOST, timeout=TIMEOUT)


def classify_one(client, text, labels, shots=None):
    """判定一条文本。返回 (类别 或 None, 依据, 原始返回)。"""
    messages = build_messages(text, labels, shots=shots)
    last_err = ""
    for attempt in range(RETRY + 1):
        try:
            resp = client.chat(model=MODEL, messages=messages)
            raw = resp["message"]["content"]
        except Exception as exc:                      # 网络/超时/服务没起来
            last_err = "%s: %s" % (type(exc).__name__, exc)
            if attempt < RETRY:
                time.sleep(2 ** attempt)              # 指数退避，别原地猛打
                continue
            return None, "call_failed", last_err

        label, why = parse_label(raw, labels)
        if label is not None:
            return label, why, raw
        # 解析不出来也值得再试一次：温度不为 0 时下一次可能就规矩了
        if attempt < RETRY:
            continue
        return None, why, raw
    return None, "call_failed", last_err


def classify_many(texts, labels=None, shots=None):
    labels = labels or evalset.labels()
    client = _client()
    results = []
    for i, text in enumerate(texts, 1):
        with console.status("[bold bright_green]判定第 %d/%d 条…" % (i, len(texts))):
            label, why, raw = classify_one(client, text, labels, shots)
        results.append({"text": text, "label": label, "why": why, "raw": raw})
    return results


def main():
    labels = evalset.labels()
    texts = [
        "今日，央行发布公告宣布降低利率，以刺激经济增长。这一降息举措将影响贷款利率，并在未来几个季度内对金融市场产生影响。",
        "本公司宣布成功收购一家在创新科技领域领先的公司，这一战略性收购将有助于公司拓展技术能力和加速产品研发。",
        "公司资产负债表显示，公司偿债能力强劲，现金流充足，为未来投资和扩张提供了坚实的财务基础。",
        "最新的分析报告指出，可再生能源行业预计将在未来几年经历持续增长，投资者应该关注这一领域的投资机会",
    ]

    print("[bold]模型[/bold] %s   [bold]地址[/bold] %s" % (MODEL, HOST))
    results = classify_many(texts, labels, shots=CLASS_EXAMPLES)

    bad = 0
    for r in results:
        head = r["text"][:28] + "…"
        if r["label"] is None:
            bad += 1
            print(">>> [bold bright_red]判不了[/bold bright_red]（%s） %s" % (r["why"], head))
            print("    原始返回：%r" % (str(r["raw"])[:120],))
        else:
            print(">>> [bold bright_green]%s[/bold bright_green]（%s） %s"
                  % (r["label"], r["why"], head))
    print("\n共 %d 条，%d 条收敛到类别体系内，%d 条判不了。"
          % (len(results), len(results) - bad, bad))


if __name__ == "__main__":
    main()

# -*- coding: utf-8 -*-
"""在评测集上跑一版提示词，把准确率算出来。

这是把前面几个零件接起来的那一步：
    evalset（标准答案） → prompt_render（渲染 messages）
        → finance_classify（调模型 + 输出校验） → eval_scorer（算分）

用法：
    python3 run_eval.py v1        # 只有角色和类别
    python3 run_eval.py v2        # 加一句输出格式约束
    python3 run_eval.py v3        # 再补四条示例
    python3 run_eval.py v4        # 再补类别边界说明

跑完会把 (sid, 标准答案, 预测) 三列写进 CSV，方便回看是哪几条错了。
依赖 ollama 与本地已拉好的模型；没有推理环境时先跑 eval_scorer.py，
它用同样的打分逻辑，不需要模型。
"""
import csv
import os
import sys

import evalset
from eval_scorer import report, score
from output_guard import parse_label
from prompt_render import CLASS_EXAMPLES, build_messages

OUT_DIR = os.environ.get("FIN_EVAL_OUT", ".")

# 四版提示词的差异全部收敛在这张表里，一版只比上一版多一个键
VARIANTS = {
    "v1": dict(shots=None, with_format_rule=False, with_boundary=False),
    "v2": dict(shots=None, with_format_rule=True, with_boundary=False),
    "v3": dict(shots=CLASS_EXAMPLES, with_format_rule=True, with_boundary=False),
    "v4": dict(shots=CLASS_EXAMPLES, with_format_rule=True, with_boundary=True),
}


def run(version):
    from finance_classify import MODEL, _client, RETRY  # 复用同一套调用与重试

    if version not in VARIANTS:
        raise SystemExit("没有这一版：%s，可选 %s" % (version, list(VARIANTS)))

    labels = evalset.labels()
    evalset.self_check()
    client = _client()
    kw = VARIANTS[version]

    pairs, rows = [], []
    parsed = 0
    samples = evalset.samples()
    for i, (sid, text, gold, hard) in enumerate(samples, 1):
        messages = build_messages(text, labels, **kw)
        raw, pred, why = "", None, "call_failed"
        for attempt in range(RETRY + 1):
            try:
                raw = client.chat(model=MODEL, messages=messages)["message"]["content"]
            except Exception as exc:
                why = "%s: %s" % (type(exc).__name__, exc)
                continue
            pred, why = parse_label(raw, labels)
            break
        parsed += int(pred is not None)
        pairs.append((sid, gold, pred))
        rows.append({"sid": sid, "hard": int(hard), "gold": gold,
                     "pred": pred or "", "why": why,
                     "raw": str(raw).replace("\n", " ")[:200]})
        print("  [%2d/%2d] %s 标准=%-6s 预测=%-6s (%s)"
              % (i, len(samples), sid, gold, pred or "判不了", why))

    res = score(pairs, labels)
    parse_rate = parsed / float(len(samples))

    path = os.path.join(OUT_DIR, "eval_%s.csv" % version)
    with open(path, "w", encoding="utf-8-sig", newline="") as f:
        w = csv.DictWriter(f, fieldnames=["sid", "hard", "gold", "pred", "why", "raw"])
        w.writeheader()
        w.writerows(rows)

    print("")
    print(report(res, labels, "%s / %s" % (version, MODEL)))
    print("\n  可解析率   : %.1f%%（模型返回能收敛成枚举值的比例）" % (parse_rate * 100))
    print("  明细已写入 : %s" % path)
    print("\n  把准确率与可解析率填进 iterate_log.py，再跟上一版比。"
          "两版之间只许差一处，否则涨跌无法归因。")
    return res, parse_rate


if __name__ == "__main__":
    run(sys.argv[1] if len(sys.argv) > 1 else "v2")

# -*- coding: utf-8 -*-
"""大模型文本分类骨架：换个行业、换个类别体系，改五处 TODO 就能用。

这份骨架把「任务定义 → 提示词设计 → 评测集 → 迭代优化」四步固化成了代码结构：
    LABELS / DESCRIPTIONS   任务定义与类别边界
    build_messages()        提示词设计
    EVALSET + evaluate()    评测集与打分
    VERSIONS + diff_once()  迭代优化，一次只改一处

纯标准库，不接模型也能跑：它会用一个假的 predict_fn 把整条链路走通，
证明「输入构造、输出校验、打分」三段都是对的，再去接真实推理。

用法：
    python3 llm_classify_skeleton.py       # 跑自检 + 演示
"""
import re

# ---------------------------------------------------------------- 任务定义
# TODO 1：换成你的类别体系。三条硬要求：
#   · 类别之间互斥，一条文本只应落进一个类别；
#   · 类别名在提示词、示例、评测集、代码里**逐字一致**，不许出现同义写法；
#   · 类别数先控制在 10 个以内，超了先考虑分层判定。
LABELS = ["新闻报道", "财务报告", "公司公告", "分析师报告"]

# TODO 2：写清每个类别的边界。这段会拼进提示词，也是标注员的标注手册。
#         判据：两个标注员照着它标同一批数据，结论应当一致。
DESCRIPTIONS = {
    "新闻报道": "第三方在陈述已经发生的事",
    "财务报告": "企业自己在披露经营与财务数据",
    "公司公告": "企业自己在宣布一件具体事项",
    "分析师报告": "第三方在给出判断与预期",
}

# TODO 3：每个类别给一条示例。示例要挑典型的，不要挑边界样本。
SHOTS = {
    "新闻报道": "今日，股市经历了一轮震荡，受到宏观经济数据和全球贸易紧张局势的影响。",
    "财务报告": "本公司年度财务报告显示，去年公司实现了稳步增长的盈利。",
    "公司公告": "本公司宣布成功完成最新一轮并购交易，收购了一家人工智能公司。",
    "分析师报告": "最新的行业分析报告指出，科技创新将成为未来增长的主要推动力。",
}

ASK = "“{text}”是 {labels} 里的什么类别？"


# ------------------------------------------------------------ 提示词设计
def build_messages(text, shots=None, with_format_rule=True, with_boundary=False):
    """渲染 messages。每个开关对应一处可独立开合的改动。"""
    sys_parts = ["现在你是一个文本分类器，你需要按照要求将我给你的句子分类到：%s类别中。" % LABELS]
    if with_format_rule:
        sys_parts.append("只输出类别名本身，不要输出任何解释、理由或标点。")
    if with_boundary:
        sys_parts.append("判定依据是文字的体裁：" +
                         "；".join("%s是%s" % (k, v) for k, v in DESCRIPTIONS.items()) + "。")
    msgs = [{"role": "system", "content": "".join(sys_parts)}]
    for label, example in (shots or {}).items():
        msgs.append({"role": "user", "content": ASK.format(text=example, labels=LABELS)})
        msgs.append({"role": "assistant", "content": label})
    msgs.append({"role": "user", "content": ASK.format(text=text, labels=LABELS)})
    return msgs


# ------------------------------------------------------------ 输出校验
_THINK = re.compile(r"<think>.*?</think>", re.S)
_FENCE = re.compile(r"```[a-zA-Z]*\s*(.*?)\s*```", re.S)
_TRIM = "\"'“”‘’ 。．.，,：:；;！!？?\n\t "


def parse_label(raw):
    """自由文本 → 枚举值 或 None。表外、模糊一律 None，绝不猜。"""
    text = _THINK.sub("", str(raw or ""))
    m = _FENCE.search(text)
    if m:
        text = m.group(1)
    text = text.strip(_TRIM)
    if text in LABELS:
        return text, "exact"
    hit = [lab for lab in LABELS if lab in text]
    if len(hit) == 1:
        return hit[0], "unique"
    if len(hit) > 1:
        return None, "ambiguous"
    return None, "out_of_scope"


# ------------------------------------------------------------ 评测集
# TODO 4：换成你自己的评测集，并且**在动提示词之前就把它标好**。
#         每类不少于 25 条、类别均衡、边界样本单独标记，是能看出差别的起点。
EVALSET = [
    ("S01", "今日，央行发布公告宣布降低利率，以刺激经济增长。", "新闻报道"),
    ("S02", "公司资产负债表显示，公司偿债能力强劲，现金流充足。", "财务报告"),
    ("S03", "本公司董事会决议通过，聘任新任首席财务官。", "公司公告"),
    ("S04", "我们维持对该板块的中性评级，认为估值已部分反映预期。", "分析师报告"),
]


def evaluate(predict_fn, **kw):
    """predict_fn(messages) -> 模型返回的原始字符串。"""
    right = parsed = 0
    detail = []
    for sid, text, gold in EVALSET:
        raw = predict_fn(build_messages(text, **kw))
        pred, why = parse_label(raw)
        parsed += int(pred is not None)
        right += int(pred == gold)
        detail.append((sid, gold, pred, why))
    n = float(len(EVALSET))
    return {"accuracy": right / n, "parse_rate": parsed / n, "detail": detail}


# ------------------------------------------------------------ 迭代优化
# TODO 5：每加一版，只准比上一版多改一个键。
VERSIONS = {
    "v1": dict(shots=None, with_format_rule=False, with_boundary=False),
    "v2": dict(shots=None, with_format_rule=True, with_boundary=False),
    "v3": dict(shots=SHOTS, with_format_rule=True, with_boundary=False),
    "v4": dict(shots=SHOTS, with_format_rule=True, with_boundary=True),
}


def diff_once(a, b):
    """相邻两版必须只差一个键，差多了当场抛错。"""
    keys = set(a) | set(b)
    changed = sorted(k for k in keys if a.get(k) != b.get(k))
    if len(changed) != 1:
        raise AssertionError("一次改了 %d 处：%s" % (len(changed), "、".join(changed)))
    return changed[0]


def _self_check():
    assert len(LABELS) == len(set(LABELS)), "类别名重复"
    assert set(DESCRIPTIONS) == set(LABELS), "边界说明与类别体系对不上"
    assert set(SHOTS) <= set(LABELS), "示例答案跑到类别体系外面去了"
    for _sid, _t, gold in EVALSET:
        assert gold in LABELS, "评测集里的标准答案 %r 不在类别体系里" % gold
    order = ["v1", "v2", "v3", "v4"]
    steps = [diff_once(VERSIONS[a], VERSIONS[b]) for a, b in zip(order, order[1:])]
    assert steps == ["with_format_rule", "shots", "with_boundary"], steps
    return steps


def _demo():
    steps = _self_check()
    print("【骨架自检通过】每一版只改一处：%s" % " → ".join(steps))

    # 一个假模型：v1 那样没有格式约束时啰嗦，加了约束就只回类别名。
    # 它的作用是把链路走通，不是模拟真实模型的判断力。
    def fake_predict(messages):
        rule_on = "只输出类别名本身" in messages[0]["content"]
        target = messages[-1]["content"]
        guess = next((lab for lab in LABELS if lab[:2] in target), LABELS[0])
        for sid, text, gold in EVALSET:
            if text in target:
                guess = gold
                break
        return guess if rule_on else "这段文字属于%s类别，因为它的表述方式符合该体裁。" % guess

    print("\n【假模型跑两版，看链路通不通】")
    for name in ("v1", "v2"):
        r = evaluate(fake_predict, **VERSIONS[name])
        print("  %s  可解析率 %.0f%%   准确率 %.0f%%"
              % (name, r["parse_rate"] * 100, r["accuracy"] * 100))
        for sid, gold, pred, why in r["detail"]:
            print("      %s 标准=%-6s 预测=%-6s (%s)" % (sid, gold, pred or "判不了", why))

    print("\n  接真实模型时，把 fake_predict 换成一次真实调用即可，"
          "其余三段（渲染、校验、打分）一行都不用动。")

    try:
        diff_once(VERSIONS["v2"], dict(VERSIONS["v4"]))
    except AssertionError as e:
        print("\n【跨版直接跳会被拦下】%s" % e)


if __name__ == "__main__":
    _demo()