15 基础文本分析

# 导入正则表达式工具，按确定的连字号规则从年报节选取词。
import re
# 导入词频计数器，使 TF 表与清洗序列共用同一输入。
from collections import Counter
# 确定印刷页 40 的句首节选，不将省略的后文纳入计数。
alibaba_report_excerpt = "Our data centers employ leading technologies in distributed fault-tolerant architecture, advanced in-house power and cooling equipment, AI-driven intelligent monitoring and operational technology,"
# 固定只删除本例明示的三个功能词，避免静默丢弃内容词。
alibaba_stopwords = {"our", "in", "and"}
# 固定六个教学词形映射，其余词元保持原样。
alibaba_lemma_map = {"centers": "center", "leading": "lead", "technologies": "technology", "distributed": "distribute", "cooling": "cool", "monitoring": "monitor"}
# 按“字母串+可选词内连字号”分词并统一转为小写。
alibaba_raw_tokens = re.findall(r"[A-Za-z]+(?:-[A-Za-z]+)*", alibaba_report_excerpt.lower())
# 先移除确定停用词，再对剩余 unigram 应用显式词形映射。
alibaba_clean_tokens = [alibaba_lemma_map.get(token, token) for token in alibaba_raw_tokens if token not in alibaba_stopwords]
# 从清洗序列直接生成词频，禁止另手工填写一套 TF 口径。
alibaba_term_frequency = Counter(alibaba_clean_tokens)
# 确定几何示例的五维词典，首维直接使用实际词元 `ai-driven`。
similarity_vocabulary = ["ai-driven", "data", "cloud", "risk", "server"]
# 按确定词典顺序从同一 TF 计数生成阿里巴巴文档向量。
alibaba_document_vector = [alibaba_term_frequency[term] for term in similarity_vocabulary]
# 断言内容词、连字号词元与重复词的计数都与课件说明一致。
assert alibaba_term_frequency["advanced"] == 1 and alibaba_term_frequency["in-house"] == 1 and alibaba_term_frequency["technology"] == 2
# 断言后续五维向量确由确定 unigram 口径生成。
assert alibaba_document_vector == [1, 1, 0, 0, 0]

# 为“代码演示：使用`scikit-learn`计算TF-IDF”，导入 TF–IDF 向量器，把金融文本转换为词项权重矩阵。
from sklearn.feature_extraction.text import TfidfVectorizer
# 为“代码演示：使用`scikit-learn`计算TF-IDF”导入 `pandas` 并绑定 `pd`，用于整理当前任务的表格、字段与时间索引。
import pandas as pd
import numpy as np  # 验证默认 L2 归一化与边界词 IDF

# 固定第一段短文本，作为 TF-IDF 文档—词项矩阵的第一行。
doc_A = 'company revenue growth is strong'
# 固定英文诉讼风险句子 `doc_B`，提供只在该文档出现的风险词。
doc_B = 'the company faces litigation risk'
# 固定英文产品收入句子 `doc_C`，与文档 A 共享 growth、strong 和 revenue。
doc_C = 'company strong growth in product revenue'
# 将三段固定短文本组成语料列表，保持文档顺序以解释后续 TF-IDF 矩阵的行索引。
corpus = [doc_A, doc_B, doc_C]

# 建立英文 TF-IDF 向量器并移除英语停用词，固定文档—词项权重口径。
# stop_words='english' 会自动移除英文停用词 (如 'is', 'the', 'in')
vectorizer = TfidfVectorizer(
    stop_words='english', use_idf=True, smooth_idf=True,
    sublinear_tf=False, norm='l2'
)

# 从三段语料学习词表与 IDF，并生成每篇文档的稀疏 TF-IDF 向量。
# fit: 学习词汇表和IDF权重
# transform: 将文档转换为TF-IDF向量
X = vectorizer.fit_transform(corpus)

# 将稀疏 TF-IDF 矩阵物化为带词项列名和文档行名的教学表格。
df_tfidf = pd.DataFrame(
    # 把稀疏文本矩阵转为教学表格可显示的稠密数组。
    X.toarray(),
    # 指定 `columns` 为输出列名，细化“代码演示：使用scikit-learn计算TF-IDF”的输出。
    columns=vectorizer.get_feature_names_out(),
    # 用文档 A、B、C 标记三行 TF-IDF 向量，便于对应原始文本。
    index=['文档 A', '文档 B', '文档 C']
)
# 标出随后展示的是移除停用词后的文档—词项 TF–IDF 权重表。
print("--- TF-IDF 权重矩阵 (已移除停用词) ---")
# 输出词项—文档 TF–IDF 表，比较三个金融句子的区分性词权重。
display(df_tfidf.round(2))  # 展示每篇文档的词频逆文档频率权重
company_idf = vectorizer.idf_[vectorizer.vocabulary_['company']]  # 取出全语料共同词的平滑逆文档频率
np.testing.assert_allclose(company_idf, 1.0)  # 验证共同词的默认平滑 IDF 等于一
np.testing.assert_allclose(np.linalg.norm(X.toarray(), axis=1), 1.0)  # 验证每篇文档向量完成 L2 归一化
top_terms = df_tfidf.apply(lambda row: row.nlargest(2).index.tolist(), axis=1)  # 提取每篇文档权重最高的两个词
display(top_terms.rename('每篇权重最高的两个词'))  # 核对每篇文档的最高权重词

# 导入 jieba，用于把机制文本切分为词元并执行手工情感词表匹配。
import jieba
# 导入标准数学函数接口供课堂扩展；本段可执行的词典计数核心不调用该模块。
import math
# 为“招行年报分析步骤1：准备数据与函数”导入 `pandas` 并绑定 `pd`，用于整理当前任务的表格、字段与时间索引。
import pandas as pd

# 固定正负面手工词表与机制文本，限定词典计数示例的输入边界。
# 这是一个简化的模拟词典，用于教学演示
posi_words = ['优势', '拓展', '积极', '增长', '机会', '收益', '专业优势']
# 固定负面词表及词序，作为手工机制示例的可检查匹配口径。
nega_words = ['萎缩', '风险', '亏损', '走势', '收窄', '违约']

# 招行年报文本
text_cmb = '本集团在持续为客户提供汇率类产品衍生交易服务的同时，发挥在利率互换等金融市场衍生交易方面的专业优势，拓展新的对客衍生交易服务品种，并积极为客户提供线上交易服务，批发客户数量和交易规模继续保持增长。'

# 定义手工词表情感函数，返回正负命中数、净计数和归一化极性。
def sentiment_analysis(text, posi_list, nega_list):
    # 用函数文档字符串说明“对输入文本进行简单的情感分析”。
    """对输入文本进行简单的情感分析"""
    # 使用jieba分词
    tokens = jieba.lcut(text)

    # 统计正面和负面词汇的数量
    posi_count = sum(1 for word in tokens if word in posi_list)
    # 统计分词结果中命中负面词表的次数，作为净计数和极性分母的另一组成项。
    nega_count = sum(1 for word in tokens if word in nega_list)

    # 计算两种情感得分
    # 简单得分
    simple_score = posi_count - nega_count

    # 情感极性得分，归一化到[-1, 1]之间
    total_senti_words = posi_count + nega_count
    # 计算 `polarity_score`（评分），量化“招行年报分析步骤1：准备数据与函数”的模型表现。
    polarity_score = (posi_count - nega_count) / total_senti_words if total_senti_words > 0 else 0

    # 返回正负命中次数、净计数与归一化极性四项指标，供调用方表格化解释。
    return {
        # 返回正负命中次数与净计数，保留原始词典计数口径。
        '正面词数量': posi_count, '负面词数量': nega_count,
        # 返回净计数与归一化极性，供透明编码检查。
        '简单得分': simple_score, '情感极性': polarity_score
    }

# 对固定机制文本执行词表情感函数，生成可表格化的四项结果。
results = sentiment_analysis(text_cmb, posi_words, nega_words)
# 核对本示例的实际分词边界，防止把连续短语误写成命中词。
cmb_tokens = jieba.lcut(text_cmb)
# 按玩具词典顺序提取实际命中的正面 token。
matched_positive_tokens = [token for token in cmb_tokens if token in posi_words]
# 确定当前 jieba 输出，避免讲解再次把“优势”写成“专业优势”。
assert matched_positive_tokens == ['优势', '拓展', '积极', '增长']

# 使用Pandas Series美化输出，并以单次打印核对词典法结果序列。
print("--- 招商银行年报情感分析结果 ---\n" + pd.Series(results).to_string())
# 输出实际命中词，令讲解直接接受可执行结果检查。
print({'命中正面词': matched_positive_tokens})

from pathlib import Path  # 管理公开审计公告结构化数据版本
from urllib.request import urlretrieve  # 复用本章隐藏设置单元安装的浏览器标识下载器
import hashlib  # 生成固定文本夹具的内容哈希
import pandas as pd  # 构造公司财年文本面板
# 按 Linux 共享数据、Windows 共享数据、项目缓存的顺序选择审计意见文件。
audit_text_path = next((candidate_path for candidate_path in [Path('/home/ubuntu/r2_data_mount/data/stock/audit_opinion.h5'), Path('C:/qiufei/data/stock/audit_opinion.h5'), Path('data/course/audit_opinion.h5')] if candidate_path.exists()), Path('data/course/audit_opinion.h5'))
if not audit_text_path.exists():
    audit_text_path.parent.mkdir(parents=True, exist_ok=True)
    urlretrieve('https://assets.qiufei.site/data/stock/audit_opinion.h5', audit_text_path)
audit_raw = pd.read_hdf(audit_text_path, key='audit_opinion')  # 读取并保留全部披露版本
audit_raw['info_date'] = pd.to_datetime(audit_raw['info_date'])  # 统一公告日期类型
audit_raw = audit_raw.query("type == 'financial_statements' and quarter.str.endswith('q4')", engine='python').copy()  # 仅保留年度财报审计公告
audit_raw['year'] = audit_raw['quarter'].str[:4].astype(int)  # 提取当期财年
opinion_words = {'unqualified': '标准无保留意见', 'unqualified_with_explanation': '带说明段无保留意见', 'qualified': '保留意见', 'disclaimer': '无法表示意见', 'adverse': '否定意见'}  # 固定意见类型中文映射
def audit_asof(fiscal_year, cutoff):  # 每个截止日从原始版本重建
    all_versions = audit_raw.loc[audit_raw['year'].eq(fiscal_year)].copy()  # 保留该财年全部版本
    eligible = all_versions.loc[all_versions['info_date'].le(cutoff)].copy()  # 先按可得日过滤
    if eligible.empty:  # 空窗口显式失败
        raise RuntimeError(f'{fiscal_year}@{cutoff.date()} 空窗')  # 点时样本缺失时立即停止而不回填未来文本
    snap = eligible.sort_values(['order_book_id', 'year', 'info_date']).drop_duplicates(['order_book_id', 'year'], keep='last')  # 再选当时最后版本
    flow = {'raw': len(all_versions), 'eligible': len(eligible), 'excluded': len(all_versions) - len(eligible), 'companies': snap.order_book_id.nunique()}  # 样本数量变化
    return snap, flow  # 返回点时数据版本

def build_audit_year(year):  # 构造当期文本与下一财年标签
    prediction_cutoff, scoring_cutoff = pd.Timestamp(f'{year + 1}-06-30'), pd.Timestamp(f'{year + 2}-06-30')  # 确定双截止日
    features, feature_flow = audit_asof(year, prediction_cutoff)  # 预测日文本版本
    label_all = audit_raw.loc[audit_raw['year'].eq(year + 1)].copy()  # 标签版本先不去重
    assert len(label_all) > 0 and label_all.info_date.min() > prediction_cutoff  # 目标不得提前可见
    labels, label_flow = audit_asof(year + 1, scoring_cutoff)  # 评分日标签版本
    features['document'] = '审计机构 ' + features.audit_agency.fillna('未披露机构') + ' 审计意见 ' + features.opinion_type.map(opinion_words).fillna('其他意见')  # 仅用特征数据版本构文档
    labels = labels[['order_book_id', 'info_date', 'opinion_type']].rename(columns={'info_date': 'label_info_date'})  # 保留标签披露日
    labels['next_modified'] = labels.opinion_type.ne('unqualified').astype(int)  # 构造下一年标签
    panel = features.rename(columns={'info_date': 'feature_info_date'}).merge(labels, on='order_book_id').assign(prediction_cutoff=prediction_cutoff, scoring_cutoff=scoring_cutoff, next_year=year + 1)  # 合并点时样本
    assert len(panel) > 0 and panel.feature_info_date.le(prediction_cutoff).all() and panel.label_info_date.gt(prediction_cutoff).all() and panel.label_info_date.le(scoring_cutoff).all()  # 日期不变量
    return panel, {'year': year, 'feature': feature_flow, 'label': label_flow, 'merged': len(panel)}  # 返回面板与样本数量变化

audit_parts, audit_flows = [], []  # 收集逐年点时样本
for fiscal_year in range(2016, 2023):  # 固定建模财年
    fiscal_panel, fiscal_flow = build_audit_year(fiscal_year)  # 从原始版本重建
    audit_parts.append(fiscal_panel)  # 保存样本
    audit_flows.append(fiscal_flow)  # 保存检查结果
audit_text_panel = pd.concat(audit_parts, ignore_index=True)  # 合并点时面板
assert not audit_text_panel.duplicated(['order_book_id', 'year']).any()  # 公司财年唯一
assert audit_text_panel.feature_info_date.lt(audit_text_panel.label_info_date).all()  # 特征严格早于标签
fixture_hash = hashlib.sha256('\n'.join(audit_text_panel['document']).encode('utf-8')).hexdigest()  # 记录文档顺序与内容哈希
print({'文件': audit_text_path.name, '键': 'audit_opinion', '字段': ['info_date', 'audit_agency', 'opinion_type'], '样本期': (audit_text_panel.year.min(), audit_text_panel.year.max()), '文档数': len(audit_text_panel), 'sha256': fixture_hash})  # 输出输入说明
display(pd.DataFrame(audit_flows))  # 展示每年版本选择与排除数

from sklearn.feature_extraction.text import TfidfVectorizer  # 构造中文字符级文本表示
from sklearn.linear_model import LogisticRegression  # 建立线性文本分类器
from sklearn.metrics import balanced_accuracy_score, confusion_matrix  # 报告类别均衡指标与错误类型
from sklearn.pipeline import Pipeline  # 确保词表、IDF和分类器只在训练折拟合
import numpy as np  # 构造训练期多数类预测
evaluation_rows = []  # 收集逐年扩展窗口结果
misclassified_rows = []  # 收集可供反馈的误分类记录
for validation_year in [2020, 2021, 2022]:  # 按预设三个年份扩展验证
    fit_cutoff = pd.Timestamp(f'{validation_year + 1}-06-30')  # 确定验证年预测日
    training_text = audit_text_panel.query('year < @validation_year and label_info_date <= @fit_cutoff')  # 只用拟合日已评分历史标签
    validation_text = audit_text_panel.query('year == @validation_year')  # 固定当年结构化公告文本
    if training_text.empty or validation_text.empty or training_text.next_modified.nunique() < 2 or validation_text.next_modified.nunique() < 2:  # 空窗或单类显式失败
        raise RuntimeError(f'{validation_year} 点时窗口为空或单类')  # 单类窗口无法形成有效分类验证
    text_pipeline = Pipeline([('tfidf', TfidfVectorizer(analyzer='char', ngram_range=(2, 3), min_df=3, max_features=1000)), ('classifier', LogisticRegression(max_iter=500, class_weight='balanced', random_state=42))])  # 折内绑定文本表示与线性分类器
    text_pipeline.fit(training_text['document'], training_text['next_modified'])  # 只在当前扩展训练窗估计词表与权重
    validation_prediction = text_pipeline.predict(validation_text['document'])  # 生成下一财年非标准意见预测
    majority_label = int(training_text['next_modified'].mode().iloc[0])  # 仅由训练折确定多数类
    majority_prediction = np.repeat(majority_label, len(validation_text))  # 建立同窗多数类基线
    fold_matrix = confusion_matrix(validation_text['next_modified'], validation_prediction, labels=[0, 1])  # 固定二乘二错误矩阵
    assert fold_matrix.shape == (2, 2) and training_text.label_info_date.le(fit_cutoff).all() and validation_text.feature_info_date.le(fit_cutoff).all() and validation_text.label_info_date.gt(fit_cutoff).all()  # 确定折内日期与矩阵不变量
    evaluation_rows.append({'year': validation_year, 'fit_cutoff': fit_cutoff.date(), 'train_n': len(training_text), 'valid_n': len(validation_text), 'train_classes': training_text.next_modified.value_counts().to_dict(), 'valid_classes': validation_text.next_modified.value_counts().to_dict(), 'model_balanced_accuracy': balanced_accuracy_score(validation_text['next_modified'], validation_prediction), 'majority_balanced_accuracy': balanced_accuracy_score(validation_text['next_modified'], majority_prediction), 'confusion_matrix': fold_matrix.tolist()})  # 保存年度指标与类别流
    errors = validation_text.loc[validation_prediction != validation_text['next_modified'], ['order_book_id', 'year', 'feature_info_date', 'label_info_date', 'document', 'next_modified']].copy()  # 定位具名公司财年错误案例
    errors['prediction'] = validation_prediction[validation_prediction != validation_text['next_modified'].to_numpy()]  # 附加错误预测类别
    misclassified_rows.append(errors.head(1))  # 每年最多保留一个具体误分类
text_evaluation = pd.DataFrame(evaluation_rows)  # 汇总年度稳定性结果

display(text_evaluation[['year', 'train_n', 'valid_n', 'model_balanced_accuracy', 'majority_balanced_accuracy']].round(4))  # 展示紧凑逐年指标
print({'逐年混淆矩阵[0,1]': dict(zip(text_evaluation.year, text_evaluation.confusion_matrix))})  # 另行保留固定标签顺序的错误流
text_errors = pd.concat(misclassified_rows, ignore_index=True)  # 合并逐年代表性误分类
stable_improvement = bool((text_evaluation['model_balanced_accuracy'] > text_evaluation['majority_balanced_accuracy']).all())  # 检查是否每年都胜多数类
print({'每年均胜多数类': stable_improvement, '结论': '仅限下一财年审计意见风险' if stable_improvement else '证据不稳定，不宣称增量价值'})  # 给出与实际结果绑定的反馈

display(text_errors[['order_book_id', 'year', 'document', 'next_modified', 'prediction']].rename(columns={
    'order_book_id': '证券代码', 'year': '财年', 'document': '文本',
    'next_modified': '实际类别', 'prediction': '预测类别'
}).head(3))