17 处理另类数据：图像与声音

from pathlib import Path
from urllib.request import urlretrieve  # 复用本章隐藏设置单元安装的浏览器标识下载器
import torch  # 运行预训练目标检测推理
import numpy as np  # 计算边界框交并比
import pandas as pd  # 整理原始预测检查表
from PIL import Image  # 读取固定 RGB 图像
from torchvision.models.detection import FasterRCNN_MobileNet_V3_Large_320_FPN_Weights, fasterrcnn_mobilenet_v3_large_320_fpn  # 确定轻量 Faster R-CNN 架构与权重枚举
# 优先使用仓库根目录随附图片，不存在时再复用章节缓存或公开下载。
detection_image_path = next((candidate_path for candidate_path in [Path('../profile.jpg'), Path('data/course/profile.jpg')] if candidate_path.exists()), Path('data/course/profile.jpg'))
if not detection_image_path.exists():
    detection_image_path.parent.mkdir(parents=True, exist_ok=True)
    urlretrieve('https://qiufei.site/web-slide-bigfinance/profile.jpg', detection_image_path)
detection_image = Image.open(detection_image_path).convert('RGB')  # 统一为模型需要的 RGB 图像
detection_weights = FasterRCNN_MobileNet_V3_Large_320_FPN_Weights.DEFAULT  # 使用 torchvision 明确版本的默认权重
detection_model = fasterrcnn_mobilenet_v3_large_320_fpn(weights=detection_weights).eval()  # 加载预训练检测器并关闭训练态随机行为
detection_tensor = detection_weights.transforms()(detection_image)  # 使用权重绑定的官方预处理说明
with torch.no_grad():  # 推理阶段不保留梯度图以控制内存
    detection_output = detection_model([detection_tensor])[0]  # 对单图执行真实 Faster R-CNN 推理
print({'输入': detection_image_path.name, '模型权重': str(detection_weights), '识别到的候选框数': len(detection_output['boxes'])})

manual_person_box = np.array([390.0, 490.0, 570.0, 700.0])  # 确定画面中人物的人工边界框真值
detection_threshold = 0.50  # 事先确定显示与评分阈值
detection_scores = detection_output['scores'].cpu().numpy()  # 取得原始置信度序列
detection_labels = detection_output['labels'].cpu().numpy()  # 取得原始 COCO 类别编号
detection_boxes = detection_output['boxes'].cpu().numpy()  # 取得原始预测框坐标
person_mask = (detection_labels == 1) & (detection_scores >= detection_threshold)  # 只评估阈值以上 person 检测
person_boxes = detection_boxes[person_mask]  # 收集全部候选人物框
def calculate_box_iou(predicted_box, truth_box):  # 计算单个预测框与人工真值的交并比
    intersection_min = np.maximum(predicted_box[:2], truth_box[:2])  # 定位交集左上角
    intersection_max = np.minimum(predicted_box[2:], truth_box[2:])  # 定位交集右下角
    intersection_area = np.prod(np.clip(intersection_max - intersection_min, 0, None))  # 计算非负交集面积
    predicted_area = np.prod(predicted_box[2:] - predicted_box[:2])  # 计算预测框面积
    truth_area = np.prod(truth_box[2:] - truth_box[:2])  # 计算人工框面积
    return float(intersection_area / (predicted_area + truth_area - intersection_area))  # 返回 IoU 供阈值判定
person_ious = [calculate_box_iou(person_box, manual_person_box) for person_box in person_boxes]  # 检查全部人物候选
true_positive_count = int(any(person_iou >= 0.5 for person_iou in person_ious))  # 以 IoU 不低于 0.5 定义命中
detection_precision = true_positive_count / max(len(person_boxes), 1)  # 报告阈值后人物检测精确率
detection_recall = float(true_positive_count)  # 单一人工人物真值下召回率等于命中指示
raw_detection_table = pd.DataFrame({'label_id': detection_labels[:5], 'score': detection_scores[:5], 'x1': detection_boxes[:5, 0], 'y1': detection_boxes[:5, 1], 'x2': detection_boxes[:5, 2], 'y2': detection_boxes[:5, 3]})  # 保留前五条原始预测
print({'阈值': detection_threshold, 'person候选': len(person_boxes), '最大IoU': max(person_ious, default=0.0), '精确率': detection_precision, '召回率': detection_recall})  # 输出完整任务指标
display(raw_detection_table.round(3))  # 展示原始标签、分数与边界框

from pathlib import Path  # 管理公开 A 股分钟数据路径
from urllib.request import urlretrieve  # 复用本章已安装的浏览器标识下载器
import h5py  # 对非 Pandas HDF5 数据集做选择性切片
import matplotlib.pyplot as plt  # 将真实价格序列渲染为图像输入
# 按 Linux 共享数据、Windows 共享数据、项目缓存的顺序选择分钟行情文件。
minute_path = next((candidate_path for candidate_path in [Path('/home/ubuntu/r2_data_mount/data/stock/minbar/600276.XSHG.h5'), Path('C:/qiufei/data/stock/minbar/600276.XSHG.h5'), Path('data/course/600276.XSHG.h5')] if candidate_path.exists()), Path('data/course/600276.XSHG.h5'))
if not minute_path.exists():
    minute_path.parent.mkdir(parents=True, exist_ok=True)
    urlretrieve('https://assets.qiufei.site/data/stock/minbar/600276.XSHG.h5', minute_path)
with h5py.File(minute_path, 'r') as minute_store:  # 只读打开公开 HDF5 文件
    minute_count = minute_store['data'].shape[0]  # 获取总分钟记录数而不全量载入
    minute_slice = minute_store['data'][minute_count - 240:minute_count]  # 只读取最后 240 条固定切片
minute_close = minute_slice['close']  # 提取绘图所需收盘价字段
fig, axis = plt.subplots(figsize=(7, 3))  # 创建适合 1280×720 投影的紧凑画布
axis.plot(minute_close, color='#174F87', linewidth=1.5)  # 绘制真实公开分钟价格路径
axis.set(title='恒瑞医药：固定 240 分钟切片', xlabel='分钟顺序', ylabel='价格')  # 标注输入对象与坐标含义
plt.tight_layout()  # 收紧边界避免图形压住页脚
plt.show()  # 输出单幅可检查的图像输入
print({'文件': minute_path.name, '数据集': 'data', '总记录': minute_count, '切片记录': len(minute_slice)})  # 输出数据来源与样本数量变化

import numpy as np  # 将真实分钟价格窗口栅格化为灰度图
import pandas as pd  # 形成最终测试期人工检查表
import torch  # 用固定PyTorch版本训练微型卷积网络
from sklearn.metrics import balanced_accuracy_score, confusion_matrix  # 比较CNN与多数类基线
window_length = 60  # 固定每幅图包含60分钟
window_count = min(240, minute_count // window_length)  # 使用数据版本末端最多240个不重叠窗口
with h5py.File(minute_path, 'r') as minute_store:  # 再次只读打开固定公开数据版本
    chart_slice = minute_store['data'][minute_count - window_count * window_length:minute_count]  # 选择性读取建模区间
price_windows = chart_slice['close'].reshape(window_count, window_length)  # 按时间顺序切成固定窗口
price_minimums = price_windows.min(axis=1, keepdims=True)  # 计算每窗绘图下界
price_ranges = np.maximum(price_windows.max(axis=1, keepdims=True) - price_minimums, 1e-8)  # 防止平价窗口除零
pixel_rows = np.rint((1 - (price_windows - price_minimums) / price_ranges) * 15).astype(int)  # 映射到16行像素坐标
chart_images = np.zeros((window_count, 16, window_length), dtype=np.float32)  # 创建固定灰度图张量
chart_images[np.arange(window_count)[:, None], pixel_rows, np.arange(window_length)[None, :]] = 1.0  # 逐分钟点亮价格比较结果
direction_labels = (price_windows[:, -1] > price_windows[:, 0]).astype(np.int64)  # 以终点高于起点定义人工可核对标签
split_position = int(window_count * 0.8)  # 按时间最终测试最后20%图像
print({'PyTorch版本': torch.__version__, '文件': minute_path.name, '窗口数': window_count, '训练/测试': (split_position, window_count - split_position), '人工真值': '末价>初价'})  # 输出输入与标签说明

import torch.nn as nn  # 组合卷积、池化与分类层
torch.manual_seed(2026)  # 固定权重初始化与训练顺序
class PriceChartCNN(nn.Module):  # 定义仅用于当前方向识别任务的微型CNN
    def __init__(self):  # 注册固定网络结构
        super().__init__()  # 初始化PyTorch模块状态
        self.features = nn.Sequential(nn.Conv2d(1, 8, 3, padding=1), nn.ReLU(), nn.MaxPool2d(2), nn.Conv2d(8, 12, 3, padding=1), nn.ReLU(), nn.AdaptiveAvgPool2d((1, 1)))  # 提取局部线段特征
        self.classifier = nn.Linear(12, 2)  # 输出上涨/非上涨两类得分
    def forward(self, chart_batch):  # 定义图像到类别得分的前向路径
        return self.classifier(self.features(chart_batch).flatten(1))  # 汇总卷积特征并分类
chart_tensor = torch.tensor(chart_images[:, None, :, :])  # 增加单通道维度
label_tensor = torch.tensor(direction_labels)  # 转换人工真值标签
chart_model = PriceChartCNN()  # 从固定种子初始化网络
chart_optimizer = torch.optim.Adam(chart_model.parameters(), lr=0.01)  # 确定课堂最小训练配置
chart_loss = nn.CrossEntropyLoss()  # 使用二分类交叉熵
for training_epoch in range(30):  # 使用短预算完成最小验证
    chart_optimizer.zero_grad()  # 清除上一轮梯度
    training_scores = chart_model(chart_tensor[:split_position])  # 只读取较早图像训练
    training_objective = chart_loss(training_scores, label_tensor[:split_position])  # 计算训练期分类损失
    training_objective.backward()  # 反向传播卷积网络梯度
    chart_optimizer.step()  # 更新当前网络参数

chart_model.eval()  # 确定网络进入最终测试评估模式
with torch.no_grad():  # 测试阶段不更新参数
    sealed_chart_predictions = chart_model(chart_tensor[split_position:]).argmax(1).numpy()  # 生成时间较晚图像预测
sealed_chart_truth = direction_labels[split_position:]  # 读取最终测试期人工真值
majority_chart_label = int(np.bincount(direction_labels[:split_position]).argmax())  # 只由训练期确定多数类
majority_chart_predictions = np.repeat(majority_chart_label, len(sealed_chart_truth))  # 建立同窗朴素基线
cnn_balanced_accuracy = balanced_accuracy_score(sealed_chart_truth, sealed_chart_predictions)  # 计算CNN平衡准确率
baseline_balanced_accuracy = balanced_accuracy_score(sealed_chart_truth, majority_chart_predictions)  # 计算多数类平衡准确率
manual_audit = pd.DataFrame({'窗口': np.arange(split_position, min(split_position + 5, window_count)), '初价': price_windows[split_position:split_position + 5, 0], '末价': price_windows[split_position:split_position + 5, -1], '人工真值': sealed_chart_truth[:5], 'CNN预测': sealed_chart_predictions[:5]})  # 列出五窗供人眼核对标签
print({'CNN平衡准确率': cnn_balanced_accuracy, '多数类平衡准确率': baseline_balanced_accuracy, '混淆矩阵': confusion_matrix(sealed_chart_truth, sealed_chart_predictions).tolist()})  # 报告同窗任务指标
display(manual_audit.round(3))  # 展示预测与可手算真值