用深度学习和 Keras 解码宇宙信号，听起来像是要处理什么外星文明电报，但落到工程上其实很朴素：把望远镜采集到的原始数据经过变换之后交给神经网络，让模型回答“这个信号是不是真实天体信号”“是脉冲星候选体还是地面干扰”“这一段里有没有快速射电暴的特征”。这个方向适合两类读者：一类是天文数据处理相关的研究人员和工程师，另一类是学了一段时间深度学习、想找一个非猫狗分类实战项目的同学。最值得关注的不是网络结构有多新颖，而是从 FITS 文件、时间序列、频谱图到可部署分类结果的整条链路，能不能在普通电脑上完整跑起来。
1. 先搞清楚“解码宇宙信号”到底要解决什么问题
1.1 信号分类、信号检测、干扰剔除是三个不同任务
先给“解码”这个词降个温。深度学习在宇宙信号处理里承担的不是翻译某种神秘语言，而是三类相对成熟的工程任务。
第一是信号分类。比如射电望远镜扫描天空时会留下大量候选体，其中只有一小部分真的是脉冲星，需要用模型把候选天体信号和噪声区分开。这类任务通常被建模成二分类，也是最容易入门的起点。
第二是信号检测。快速射电暴这类瞬态信号持续时间极短，在长观测数据里找它就像大海捞针。检测任务关注的不是单个时刻，而是从长时间序列里定位异常片段。
第三是干扰剔除。地面雷达、手机基站、卫星通信都会进入望远镜的数据，造成射频干扰。这些干扰形态复杂，和真实天体信号混在一起时，传统阈值规则很难覆盖，于是很多项目转向用深度模型做异常识别。
还有一个容易误会的点：宇宙信号解码不等于直接理解信号的物理含义。深度学习模型给你的是“这一段疑似信号”或者“这类信号的概率”，物理结论仍然需要研究人员进一步确认。把目标设定成二分类或者异常检测，项目边界会清晰很多。
1.2 为什么不用传统规则，而要用神经网络
传统方法不是不能用。脉冲星搜寻早期有很多基于周期统计和信噪比阈值的算法，至今仍然在部分生产管线里运行。问题是真实数据里噪声分布不均匀、信号形态多变、干扰模式更新快，手写规则的维护成本非常高。深度学习的优势在于：只要提供足够的标注样本，模型可以自己学习“看起来像信号”的判别特征，不用人为枚举所有波形。
但这不代表模型可以随便乱来。深度学习在宇宙信号任务里能不能发挥作用，很大程度上取决于样本质量和标注一致性。我见过不少项目模型结构本身没问题，最后效果差是因为训练集里正样本标注来自不同人、不同观测条件，尺度没有对齐。所以进入建模之前，先确认数据准备这件事，它值得你花费一半以上的时间。
2. 环境与数据：先把数据形态对齐再谈模型
2.1 运行环境怎么准备
Keras 现在的典型用法是作为 TensorFlow 的高层接口，安装时通常直接安装 TensorFlow 即可：
python -m venv venv source venv/bin/activate # Windows 下用 venv\Scripts\activate pip install --upgrade pip pip install numpy scipy matplotlib tensorflow如果你希望使用 PyTorch 或者 JAX 作为 Keras 后端，需要安装对应框架版本，并按 Keras 3 的配置方式指定KERAS_BACKEND环境变量。初学者建议先走 TensorFlow 后端，资料最多，踩坑时最容易找到答案。
硬件上，纯 CPU 也能训练小规模的 CNN 或一维卷积模型，只是速度偏慢。以一张 256x256 的灰度频谱图为例，如果样本只有几千张，CPU 训练十几分钟到一小时是正常的。如果数据量上到十万级，或者需要频繁调参，建议准备一块至少 4GB 显存的 GPU，入门级消费卡就够用。本机没有 GPU 时，用深度学习云平台租一个带 GPU 的实例也是常见选择。
2.2 数据形态：时间序列、频谱图、瀑布图
宇宙信号数据进入 Keras 之前，常见形态有三种：
- 原始时间序列：一维数组，每个点是某个采样时刻的强度或功率。适合用一维卷积、LSTM 处理。
- 频谱图：二维数组，横轴时间、纵轴频率、颜色代表功率。本质是灰度图像，适合用 CNN。
- 瀑布图：射电天文里常把频率-时间信息画成瀑布图，和频谱图思路一致，只是数据组织方式略有差别。
实际项目里，很多团队会优先把时间序列转成频谱图，再按图像分类处理。原因有三个：一是图像输入对时间偏移有一定的鲁棒性；二是卷积网络处理二维时频模式比较成熟；三是可视化诊断方便，模型输出错了可以直接看频谱图找原因。
把一维时间序列转成频谱图可以用scipy.signal.spectrogram：
import numpy as np from scipy import signal def make_spectrogram(ts, fs=1000.0, nperseg=256): freqs, times, spec = signal.spectrogram(ts, fs=fs, nperseg=nperseg) # 功率谱动态范围很大，取对数后更适合神经网络 return np.log1p(spec)这里nperseg是每个窗口的采样点数，直接影响频率分辨率和时间分辨率的平衡。窗口越大，频率分辨率越高，但时间分辨率越低。具体取值没有统一标准，通常要结合信号持续时间和采样率来试。
2.3 公开数据从哪里开始
如果手上没有观测数据，一个比较稳妥的起点是使用公开的候选体数据集，比如 HTRU 数据集这种脉冲星候选体分类任务，在机器学习社区里已经流传很久，文件不大、标签清晰、适合验证管线。另外 SETI 类项目的公开信号数据也已经发布过，可以做信号检测或者异常识别的练习。
需要提醒的是，这类公开数据经过了一定程度清洗，拿来学习没问题，但不能直接等同于你后续真实任务的分布。真实数据里常见的通道坏点、时间戳异常、不同观测系统的格式差异，公开数据里往往已经被处理掉了。更合理的做法是：先用公开数据跑通代码，再拿自己的一小批真实数据做二次验证。
2.4 预处理和归一化的判断标准
信号数据最容易被忽略的是归一化。不同观测时间、不同望远镜的功率绝对数值可能差几个数量级，如果直接输入模型，模型很容易被整体幅度带偏。通常做法是：
- 对功率谱做对数压缩；
- 减去均值再除以标准差，或者缩放到 [0, 1] 区间；
- 对缺失值做插值或者直接掩膜，不能让
nan进入模型； - 检查类别比例，正负样本差距过大会导致训练不稳定。
归一化完毕后，还要做一个关键检查：把处理后的频谱图画出来看一遍。发现图像完全偏黑、偏白，或者某个通道出现奇怪的亮线，说明预处理参数有问题。先调整预处理，再进入模型训练，这是省时间的好习惯。
3. 用 Keras 搭第一个信号分类模型
3.1 方案一：把频谱图当图片做 CNN 分类
信号转成频谱图之后，分类任务就回到了熟悉的图像分类框架。一个基础的卷积模型可以这样写：
from tensorflow import keras from tensorflow.keras import layers model = keras.Sequential([ layers.Input(shape=(256, 256, 1)), layers.Conv2D(32, (3, 3), activation='relu', padding='same'), layers.MaxPooling2D((2, 2)), layers.Conv2D(64, (3, 3), activation='relu', padding='same'), layers.MaxPooling2D((2, 2)), layers.Conv2D(128, (3, 3), activation='relu', padding='same'), layers.GlobalAveragePooling2D(), layers.Dropout(0.3), layers.Dense(1, activation='sigmoid') ]) model.compile( optimizer=keras.optimizers.Adam(learning_rate=1e-3), loss='binary_crossentropy', metrics=['accuracy', 'precision', 'recall', 'AUC'] )这里用sigmoid做二分类输出，输出值是正样本概率。GlobalAveragePooling2D相比直接Flatten，参数更少，对过拟合更友好。Dropout(0.3)是常见正则化手段，防止模型死记训练集。
输入形状必须和预处理后的频谱图匹配。如果你的图不是 256x256，要么在预处理时统一 resize，要么把Input的宽高改成实际尺寸。不要为了套代码而硬改数据形状，而是先确认数据真实尺寸，再决定用Resizing层还是直接改输入维度。
3.2 方案二：用一维卷积处理原始时间序列
如果不想经过频谱图转换，或者信号本身较短、时间结构更重要，可以直接用一维卷积处理原始时间序列：
from tensorflow import keras from tensorflow.keras import layers model = keras.Sequential([ layers.Input(shape=(2048, 1)), layers.Conv1D(32, kernel_size=7, activation='relu', padding='same'), layers.MaxPooling1D(pool_size=2), layers.Conv1D(64, kernel_size=7, activation='relu', padding='same'), layers.GlobalAveragePooling1D(), layers.Dropout(0.3), layers.Dense(1, activation='sigmoid') ])每段输入长度必须一致。如果信号长短不一，需要先做截断或者补零。截断时要注意保留感兴趣的核心区域，不能简单从头开始切。比如瞬态信号可能出现在片段中后段，从开头截取 2048 点可能什么都没截到。更合理的做法是先通过峰值检测定位信号区域，再以信号区域为中心取固定窗口。
一维卷积的优势是更贴近原始数据，不丢失相位信息；缺点是训练更慢，对噪声更敏感，输入对齐要求更高。如果只是入门，我更推荐先把频谱图方案做通，再做一维方案对比。
3.3 核心参数怎么选
建模阶段最常被问到的就是几个参数：要训练几轮、batch 多大、学习率多少。
从经验上看，可以先给一组保守默认值，而不是一上来就拉满：
| 参数 | 保守取值 | 判断依据 |
|---|---|---|
| epochs | 30 | 看验证集 loss 是否还在下降，建议配合早停 |
| batch_size | 16 或 32 | 太大会占显存，太小训练抖动明显 |
| learning_rate | 1e-3 起步 | 观察 loss 下降速度，卡住再降到 1e-4 |
| dropout | 0.3 到 0.5 | 样本量少、过拟合明显时调大 |
| validation_split | 0.2 | 留出独立验证集，不参与训练 |
这些参数之间是联动的。batch 减小，梯度噪声变大，学习率可能也要相应调小。epoch 不是越大越好，很多模型在 10 轮之后验证集就不再提升，反而开始过拟合。所以不要盲目照搬网上的 100 轮设置，要让训练过程本身告诉你答案。
4. 训练与验证：别只看准确率
4.1 评估指标怎么选
宇宙信号数据里，“负样本远多于正样本”是常态。比如候选体列表中真正被确认的脉冲星可能只有百分之几，其他都是干扰和噪声。这种情况下准确率没有任何参考价值——模型全部预测为负样本，准确率也能到 95% 以上，但一个正样本都找不出来。
所以评估要以精确率、召回率、F1 和混淆矩阵为主：
- 精确率高，说明模型预测为正的样本里，真信号比例高。
- 召回率高，说明真实信号被漏掉的少。
- F1 是两者的平衡，适合类别不平衡场景。
- 混淆矩阵能直接看出模型在哪种类型上犯错。
在模型编译时把precision、recall、AUC加进metrics列表，训练结束后可以直接看。如果原来没有加，重新训练一次即可，成本不高。
4.2 处理类别极度不平衡
正负样本严重不平衡时，有几个常用手段。
第一，设置class_weight。Keras 的fit支持class_weight，让模型在计算 loss 时给少数类更大的权重：
class_weight = {0: 1.0, 1: 5.0} history = model.fit(train_ds, epochs=30, validation_data=val_ds, class_weight=class_weight)权重不是随便设的。常见做法是先按样本比例的反比设置一个初值，然后观察召回率是否提升、精确率是否下降太多，再微调。权重过大容易让模型疯狂预测为正样本，反而破坏精确率。
第二，做欠采样或过采样。正样本不足时，可以对多数类随机抽样，或者对少数类做增强，比如给频谱图加轻微噪声、轻微位移。增强要注意守恒，不能改变信号原有的物理特征。
第三，如果不平衡特别严重，可以考虑 focal loss 这类专门为困难样本设计的损失函数。它在目标检测领域用得很多，信号检测任务里也能借鉴，但代码需要自己实现，入门阶段不着急。
4.3 常用训练策略：早停、检查点、学习率衰减
训练宇宙信号模型，建议从第一天就把三个回调函数加上：早停、模型检查点、学习率调度。
from tensorflow.keras.callbacks import EarlyStopping, ModelCheckpoint, ReduceLROnPlateau callbacks = [ EarlyStopping(monitor='val_loss', patience=5, restore_best_weights=True), ModelCheckpoint('best_model.keras', monitor='val_loss', save_best_only=True), ReduceLROnPlateau(monitor='val_loss', factor=0.5, patience=3, min_lr=1e-6) ]早停的作用是当验证集 loss 连续多轮不再下降时自动停止，避免无效训练时间。检查点保存最佳模型，即使后面训练跑崩了，也能回到最好状态。学习率衰减是在 loss 卡住时自动把学习率缩小一半，帮助损失函数继续下降。
这三个回调都不复杂，但非常实用。我见过太多人训练结束时发现忘记保存模型，不得不从头再跑一遍。把检查点放在最前面，是成本最低的保护措施。
5. 批量预测与工程化落地
5.1 用 tf.data 构建稳定的数据管道
单个模型能跑通之后，接下来的问题是：如果有一批新观测数据要批量处理，应该怎么组织数据流。最直接的做法是用tf.data.Dataset构建管道，而不是把所有文件一次性读进内存。
import tensorflow as tf def load_and_preprocess(path, label): # 读取频谱图，归一化，转成 float32 image = tf.io.read_file(path) image = tf.image.decode_png(image, channels=1) image = tf.image.resize(image, (256, 256)) image = tf.cast(image, tf.float32) / 255.0 return image, label paths = [...] # 文件路径列表 labels = [...] # 标签列表 dataset = tf.data.Dataset.from_tensor_slices((paths, labels)) dataset = dataset.map(load_and_preprocess, num_parallel_calls=tf.data.AUTOTUNE) dataset = dataset.batch(32).prefetch(tf.data.AUTOTUNE)prefetch(tf.data.AUTOTUNE)的作用是在 GPU 计算的同时提前准备下一批数据，减少 GPU 空等时间。num_parallel_calls控制并行读取的进程数。这个管道看起来只是细节，但数据量大了之后，它往往比模型本身的瓶颈更明显。
注意：训练时用均值、标准差归一化，推理时必须使用同一套统计量，不能每次重新计算，否则训练和推理的数据分布会对不上。
5.2 批量推理、结果命名和失败重试
批量处理图像或者批量处理信号文件时，最容易被忽略的是输出文件的管理。实际落地时建议：
- 每条输入保留原始文件名，预测结果写入 CSV，至少包含文件名、预测概率、标签。
- 输出目录按日期或者任务名建子目录，避免多次运行互相覆盖。
- 处理过程中把失败样本单独记录，不要静默跳过。
- 如果任务耗时很长，考虑断点续跑：已处理的文件跳过，只处理剩余部分。
批量任务不能只看“能不能跑”，还要看“跑挂了之后怎么恢复”。我一般会先用 5 到 10 条样本跑一遍全流程，确认输入、输出、日志都正常，再放开全部文件。这个过程用不了几分钟，但能省掉大批任务跑到一半才发现命名冲突、输出路径不存在之类问题的麻烦。
5.3 模型保存与加载
训练完的模型要保存成可复用的格式。Keras 里最简单的做法是：
model.save('signal_classifier.keras') loaded_model = keras.models.load_model('signal_classifier.keras')如果要把模型接入 Web 服务或者移动端，可以考虑导出成 TensorFlow SavedModel 格式，再做后续部署。在部署阶段要特别注意浮点数格式：模型训练时一般用 FP32，推理时为了提速和节省显存，可以在支持的环境下换成 FP16 或者 BF16，但需要先做精度验证，确认换精度之后混淆矩阵没有明显变差。
这里有个常见误区：浮点数精度越低，推理越快，但模型输出会变。如果只是自己验证，FP32 够了；如果是上线服务，才需要认真对比 FP32、FP16、TF32 的实际效果。不要为了追求快，直接在所有环境里无脑开 FP16。
6. 常见报错与排查顺序
6.1 训练时 loss 为 NaN
loss 变成 NaN 是信号类模型很常见的现象，尤其是直接输入原始时间序列时。可能的原因有三个：输入数据里有 NaN 或无穷大，归一化没有做好；学习率过大导致梯度爆炸；采样率或者窗口参数导致某些样本异常。
排查顺序是：先检查数据，打印输入数组的np.isnan()和np.isinf()统计；再降低学习率到 1e-4 试试；最后检查是不是某一条样本导致的。不要一上来就换网络结构，NaN 问题绝大多数不是模型结构造成的。
6.2 显存或内存不足
频谱图数据通常不大，但批量处理时序数据时，如果一次性把几百条长序列读进内存，很容易把内存撑爆。解决方案是按需读取、用小 batch、边读边处理。如果 GPU 显存不足，调整策略是：
- 减小 batch size。
- 把频谱图尺寸缩小，比如从 256x256 降到 128x128。
- 在
tf.data管道里做实时预处理，而不是预先保存所有处理后的数组。 - 检查是否同时打开了多个训练进程，占用了同一块显存。
6.3 准确率很高但实际不能用
这个问题非常典型。训练准确率 98%，但拿到真实观测数据上预测，效果很差。原因通常不是模型过拟合，而是训练数据和真实数据分布不一致：公开数据集已经清洗过，真实数据里有坏通道、时间戳偏移、不同观测系统的校准差异。
处理思路是引入一小批真实数据做微调，或者至少做验证。哪怕只有一百条，也能很清楚地看到模型在真实数据上的表现。如果相差很大，先对比频谱图的可视化结果，确认预处理阶段是否引入了分布偏移。
6.4 通用排查链路
如果你在任何一个环节卡住，建议按这个顺序排查，而不是随手改参数：
- 先看现象：是报错、卡住、无输出，还是输出明显异常。
- 再看输入：文件路径、编码、格式、尺寸、缺失值、标签是否对齐。
- 再看环境：TensorFlow 版本、Python 版本、GPU 驱动、显存占用。
- 再看参数：batch size、学习率、验证集划分方式、回调配置。
- 最后再看模型：输入形状是否匹配、最后一层激活函数是否合理、loss 是否和任务匹配。
注意：数据问题最容易被伪装成模型问题。先把数据管线验证清楚，再调整网络结构，效率会高很多。
我个人的习惯是，每次开始一个新的数据集，都先单独写一个可视化脚本，把加载出来的频谱图连同样本标签一起保存成图片，人工过一遍。这个步骤花不了多少时间，但能过滤掉大量后面才会爆出来的低级问题。如果你准备拿一套自己的信号数据来跑，建议先别处理全量文件，先挑 50 条样本跑通全流程，确认归一化、标签、输入形状和预测输出都没问题，再放开到全部数据。很多看起来像“模型不好用”的现象，最后查下来都会落到数据管线和归一化上。把前面这些基础打牢，用 Keras 解码宇宙信号这件事，才算真正从演示走向了可用。
