近年来航班流量持续增长空域拥堵和航班延误已经成为民航运行中绕不开的难题。空管、机场、航司和飞行计划系统都有同一个需求提前知道未来一段时间某个空域会不会拥挤航班会怎么走流量会不会超过扇区负荷。这个需求背后就是“空域预测”。传统做法通常是为每个机场、每个扇区单独训练一个预测模型换一个场景就要重新标注、重新建模成本很高迁移效果不稳定。而“基础模型”的思路正好相反先在大规模、多来源的空域数据上进行预训练让模型学会空域运行的通用规律再通过少量业务数据微调适配到具体的流量预测、延误预测或扇区负荷预测任务上。本文将围绕“Foundation Model to Predict Airspace”这一主题从概念、数据构造、模型设计到预训练与微调完整演示一个最小可运行的空域预测系统。文章适合有 Python 基础、对深度学习时间序列建模感兴趣的开发者阅读。读完你不仅能理解基础模型在空域场景的应用思路还能拿到一套可以本地运行的代码示例并了解真实落地时需要注意的工程问题。1. 空域预测为什么需要基础模型1.1 空域预测到底在预测什么空域是一个连续的三维空间但实际业务中往往不会直接对连续空间建模而是把某一高度层范围内的空域切成网格每个网格单元记录一段时间内的航班数量、平均飞行速度、爬升下降状态等。这样空域预测就变成了“网格时空序列”预测问题。常见任务包括网格流量预测预测下一个时间片每个网格单元内的航班数量。扇区负荷预测预测某个管制扇区未来 15/30/60 分钟的交通负荷。航迹补全与预测根据历史轨迹预测飞机未来位置。延误传播预测某机场延误后影响如何扩散到周边空域。这些任务虽然名称不同但底层的时空特征高度相似都依赖航班位置、速度、气象、机场容量等信息都存在明显的时间和空间关联。这正是基础模型可以发挥价值的地方。1.2 什么是基础模型基础模型Foundation Model是指在大规模数据上进行预训练学习到通用领域知识再通过微调适配到多个下游任务的模型。它和传统“一个任务一个模型”的最大区别在于“通用性”。在文本领域BERT、GPT 都是典型的基础模型先在大规模语料上学习语言规律再用于分类、生成、问答等任务。在时空预测领域基础模型的思路同样成立先让模型在大规模空域运行数据上学习时空变化的通用模式比如高峰小时的流量形态、天气影响下的流量下降、机场放行率对区域流量的传导等再通过少量目标场景数据微调得到业务可用的预测模型。1.3 基础模型相对传统方案的优势传统流量预测常用 ARIMA、LSTM、CNN 或 GBDT这些模型在小规模单场景数据上可以做得不错但有几个明显限制场景迁移能力弱A 机场训练的模型很难直接用于 B 机场。特征工程成本高每个场景都要人工选择特征。长距离时空依赖建模困难传统序列模型对远距离空间交互和长周期时间依赖的捕捉有限。基础模型通过预训练和微调把大部分通用规律沉淀在模型参数中。新场景只需要少量数据即可适配同时 Transformer 结构天然适合建模长距离依赖也方便融入网格、轨迹、气象等多模态信息。1.4 基础模型在空域预测中的落地路径在空域场景中基础模型并不是“拿来即用”的通用大模型而是需要结合领域数据的“预训练微调”框架数据准备收集历史航班轨迹、航班计划、气象数据、机场运行数据。网格化处理把原始航班数据映射到统一时空网格。自监督预训练用大量未标注数据训练模型还原被遮挡的时空片段学习空域运行的通用规律。下游微调用少量标注数据训练模型预测网格流量、扇区负荷或延误。推理与部署将模型接入运行系统输出未来时刻的空域状态。接下来的章节我们就按这个路径用一份合成数据搭建一个最小可运行的演示系统。2. 技术路线从时空网格到预训练-微调2.1 数据层空域预测可用的数据源很多常见包括ADS-B / 雷达轨迹提供航班经度、纬度、高度、速度、航向等实时信息。航班计划起飞时间、降落时间、航路、机型。气象数据风场、能见度、雷暴区域。机场运行数据跑道容量、放行率、延误状态。空域结构数据扇区边界、航路点、限制区。真实项目中数据质量是最大的挑战。本文为了方便演示使用合成数据模拟航班流量在空域网格上的分布和变化。2.2 表示层把原始空域数据组织成模型能理解的张量常见有两种思路时空网格序列把空域按经纬度切成H x W的网格每个时间片生成一个(H, W)矩阵连续时间片组成(T, H, W)序列。图结构序列把航路点和扇区看作节点航路看作边用图神经网络建模节点间的流量传播。本文采用第一种方式优点是实现简单、通用性强。真实场景如果航路结构明显可以在此基础上扩展为图结构。2.3 模型层模型采用基于 Transformer 的编码器核心组件包括输入嵌入层把(H, W)网格展平成向量映射到d_model维。位置编码层给每个时间步加入时序位置信息。Transformer Encoder捕捉时间维度的依赖关系。输出投影层根据任务输出重构序列或预测未来网格。整个训练过程分为两个阶段预训练阶段随机遮挡输入序列中的部分时间步模型需要还原被遮挡的网格值这就是自监督学习。微调阶段使用未来时刻网格值作为标签让模型输出下一时刻的预测值。整体流程可以简化为历史网格序列 (T, H, W) | v 随机遮挡 - 预训练还原 - 得到通用空域模型 | v 输入最近 T 个时间片 - Transformer - 下一时刻网格预测3. 环境准备与项目结构3.1 运行环境示例代码基于 Python 和 PyTorch依赖较少适合本地学习。建议环境如下Python 3.9 或以上版本PyTorch 2.0 或以上版本NumPyPandas可选用于数据处理Matplotlib可选用于结果可视化scikit-learn可选用于指标计算安装依赖可以使用以下命令pip install torch numpy pandas matplotlib scikit-learn本示例没有使用 CUDA 也能运行只是训练速度会慢一些。如果机器有 GPUPyTorch 会自动使用 GPU 加速。3.2 项目结构为了保持代码清晰建议按下面的目录组织项目airspace_foundation/ ├── data/ │ └── synthetic_airspace.py ├── model/ │ └── airspace_transformer.py ├── train.py ├── inference.py └── checkpoints/ # 训练后保存模型权重其中data目录负责生成合成空域数据model目录定义基础模型结构train.py完成预训练和微调inference.py加载训练好的模型进行预测。4. 构造空域时序数据集4.1 为什么使用合成数据真实空域数据涉及空管系统数据使用协议、航班隐私、安全边界等问题并不是所有人都能方便获取。为了演示完整的建模流程我们先用合成数据验证方法和代码。合成数据的设计要尽量贴近真实空域的流量特点有高峰和低谷、有局部热点区域、有时间变化趋势。4.2 合成数据生成逻辑下面代码会生成一个[num_timesteps, grid_size, grid_size]的三维数组表示空域网格每个时间片的航班流密度。我们随机设置了几个热点中心每个中心附近流量较高整体流量随时间呈正弦波动。# data/synthetic_airspace.py import numpy as np def generate_airspace_data(num_timesteps1000, grid_size8, num_centers3, seed42): 生成合成的空域网格流量数据。 参数 num_timesteps: 生成的时间步数量 grid_size: 空域网格的边长 num_centers: 热点中心数量 seed: 随机种子 返回 shape 为 (num_timesteps, grid_size, grid_size) 的 float32 数组 rng np.random.default_rng(seed) # 随机生成热点中心坐标 centers rng.uniform(0, grid_size, size(num_centers, 2)) strengths rng.uniform(0.8, 2.0, sizenum_centers) data np.zeros((num_timesteps, grid_size, grid_size), dtypenp.float32) for t in range(num_timesteps): field np.full((grid_size, grid_size), 0.5, dtypenp.float32) for i in range(grid_size): for j in range(grid_size): # 取网格中心点坐标方便计算距离 x, y i 0.5, j 0.5 for (cx, cy), strength in zip(centers, strengths): dist np.sqrt((x - cx) ** 2 (y - cy) ** 2) # 热点中心附近流量更高并叠加时间变化 field[i, j] strength * np.exp(-dist) * (1 0.3 * np.sin(t / 20)) data[t] field return data def create_sequences(data, input_len24, pred_len1): 把原始网格数据切分为输入序列和预测目标。 参数 data: shape 为 (num_timesteps, grid_size, grid_size) input_len: 输入的时间步数 pred_len: 预测的未来时间步数 返回 X: shape 为 (样本数, input_len, grid_size, grid_size) Y: shape 为 (样本数, pred_len, grid_size, grid_size) X, Y [], [] for t in range(len(data) - input_len - pred_len 1): X.append(data[t:t input_len]) Y.append(data[t input_len:t input_len pred_len]) return np.stack(X), np.stack(Y)核心逻辑说明热点中心用均匀分布随机生成模拟不同空域结构下的流量密集区域。strength * np.exp(-dist)模拟距离热点越远流量越低的空间衰减。np.sin(t / 20)模拟一天或一段时间内的流量波动。create_sequences使用滑动窗口把连续时间片切成训练样本。4.3 数据归一化训练神经网络前需要对数据进行归一化否则数值范围差距过大会影响收敛。下面是一个简单的 Min-Max 归一化函数def normalize(data): 将数据缩放到 [0, 1] 区间。 min_val data.min() max_val data.max() return (data - min_val) / (max_val - min_val 1e-8)注意在真实项目中归一化参数只能在训练集上计算不能使用验证集和测试集的信息否则会造成数据泄漏。5. 实现基础模型AirspaceTransformer5.1 位置编码Transformer 本身不包含时序位置信息因此需要显式加入位置编码。常用的正弦位置编码实现如下# model/airspace_transformer.py import math import torch import torch.nn as nn class PositionalEncoding(nn.Module): 标准正弦位置编码。 def __init__(self, d_model, max_len512): super().__init__() pe torch.zeros(max_len, d_model) position torch.arange(0, max_len).unsqueeze(1).float() div_term torch.exp( torch.arange(0, d_model, 2).float() * (-math.log(10000.0) / d_model) ) pe[:, 0::2] torch.sin(position * div_term) pe[:, 1::2] torch.cos(position * div_term) self.register_buffer(pe, pe) def forward(self, x): # x: (batch, seq_len, d_model) return x self.pe[: x.size(1)]这里的register_buffer会把位置编码注册为模型参数的一部分但它不需要更新梯度模型保存和加载时会一起保存。5.2 AirspaceTransformer 模型模型的核心结构是 Transformer Encoder同时支持两种输出模式reconstruct模式输出整个输入序列每个时间步的网格值用于预训练。forecast模式只输出最后一个时间步之后的一帧预测结果用于下游微调。class AirspaceTransformer(nn.Module): 面向空域网格序列的基础模型。 参数 grid_size: 空域网格边长 d_model: Transformer 隐藏维度 nhead: 多头注意力头数 num_layers: Transformer Encoder 层数 max_len: 位置编码最大长度 def __init__(self, grid_size8, d_model64, nhead4, num_layers2, max_len512): super().__init__() self.grid_size grid_size self.d_model d_model # 输入嵌入层把每个时间步的网格展平并映射到 d_model self.input_embedding nn.Linear(grid_size * grid_size, d_model) # 位置编码 self.pos_encoder PositionalEncoding(d_model, max_len) # Transformer Encoder encoder_layer nn.TransformerEncoderLayer( d_modeld_model, nheadnhead, batch_firstTrue ) self.encoder nn.TransformerEncoder(encoder_layer, num_layersnum_layers) # 输出投影层映射回网格维度 self.output_proj nn.Linear(d_model, grid_size * grid_size) def forward(self, x, modeforecast): 参数 x: (batch, seq_len, grid_size, grid_size) mode: reconstruct 表示还原全序列 forecast 表示预测下一时刻 batch, seq_len, h, w x.shape # 把网格展平成特征向量 x x.reshape(batch, seq_len, h * w) x self.input_embedding(x) x self.pos_encoder(x) x self.encoder(x) if mode reconstruct: # 还原每个时间步的网格 out self.output_proj(x) return out.reshape(batch, seq_len, h, w) elif mode forecast: # 使用最后一个时间步的编码结果预测下一时刻 last x[:, -1, :] out self.output_proj(last) return out.reshape(batch, h, w) else: raise ValueError(fUnsupported mode: {mode})这个模型虽然规模不大但已经具备基础模型“共享参数、多任务输出”的特征。预训练阶段用reconstruct模式学习通用空域规律微调阶段用forecast模式输出下一时刻预测。5.3 为什么选择这样的模型结构选择 Transformer Encoder 而不是更复杂的 Encoder-Decoder是因为本文演示的任务是“根据历史还原未来”本质上更接近序列表示学习。Encoder 可以直接输出每个时间步的隐含表示便于同时支持重构和预测。真实项目中如果要做多步预测或生成完整轨迹可以在这个基础上扩展 Decoder 结构。6. 预训练与下游微调完整实战6.1 训练脚本总体设计train.py是整个训练流程的入口支持两个模式pretrain自监督预训练随机遮挡部分时间步让模型还原遮挡位置。finetune使用历史序列预测下一时刻加载预训练权重后微调。训练脚本中加入了随机种子控制保证实验结果可复现。# train.py import argparse import numpy as np import torch import torch.nn as nn from torch.utils.data import DataLoader, TensorDataset from data.synthetic_airspace import create_sequences, generate_airspace_data from model.airspace_transformer import AirspaceTransformer def normalize(data): min_val data.min() max_val data.max() return (data - min_val) / (max_val - min_val 1e-8) def mask_timesteps(batch, mask_ratio0.15): 随机遮挡部分时间步用于自监督预训练。 参数 batch: (batch_size, seq_len, grid_size, grid_size) mask_ratio: 遮挡比例 返回 masked_batch: 遮挡后的输入 mask: 遮挡掩码True 表示该位置被遮挡 mask torch.rand(batch.size(0), batch.size(1), 1, 1, devicebatch.device) mask mask mask_ratio mask mask.expand_as(batch) masked_batch batch.masked_fill(mask, 0.0) return masked_batch, mask def evaluate_forecast(model, loader, criterion, device): 评估预测模式下的损失和平均绝对误差。 model.eval() total_loss 0.0 total_mae 0.0 with torch.no_grad(): for xb, yb in loader: xb xb.to(device) yb yb.to(device) pred model(xb, modeforecast) total_loss criterion(pred, yb).item() * xb.size(0) total_mae (pred - yb).abs().sum().item() n len(loader.dataset) return total_loss / n, total_mae / n def main(): parser argparse.ArgumentParser(descriptionAirspace Foundation Model Training) parser.add_argument(--mode, typestr, choices[pretrain, finetune], requiredTrue) parser.add_argument(--num-timesteps, typeint, default1200) parser.add_argument(--grid-size, typeint, default8) parser.add_argument(--input-len, typeint, default24) parser.add_argument(--d-model, typeint, default64) parser.add_argument(--nhead, typeint, default4) parser.add_argument(--num-layers, typeint, default2) parser.add_argument(--batch-size, typeint, default32) parser.add_argument(--epochs, typeint, default20) parser.add_argument(--lr, typefloat, default1e-3) parser.add_argument(--mask-ratio, typefloat, default0.15) parser.add_argument(--init-ckpt, typestr, defaultNone, help微调时加载预训练模型) parser.add_argument(--save-path, typestr, defaultcheckpoints/airspace_model.pt) args parser.parse_args() torch.manual_seed(42) np.random.seed(42) device torch.device(cuda if torch.cuda.is_available() else cpu) print(fUsing device: {device}) # 1. 生成并归一化数据 raw generate_airspace_data( num_timestepsargs.num_timesteps, grid_sizeargs.grid_size, ) data normalize(raw) # 2. 划分序列数据集 X, Y create_sequences(data, input_lenargs.input_len, pred_len1) split_idx int(len(X) * 0.8) X_train, X_val X[:split_idx], X[split_idx:] Y_train, Y_val Y[:split_idx], Y[split_idx:] # 3. 包装成 PyTorch Dataset X_train_t torch.tensor(X_train, dtypetorch.float32) X_val_t torch.tensor(X_val, dtypetorch.float32) Y_train_t torch.tensor(Y_train, dtypetorch.float32).squeeze(1) Y_val_t torch.tensor(Y_val, dtypetorch.float32).squeeze(1) # 4. 初始化模型 model AirspaceTransformer( grid_sizeargs.grid_size, d_modelargs.d_model, nheadargs.nhead, num_layersargs.num_layers, ).to(device) if args.mode finetune and args.init_ckpt is not None: ckpt torch.load(args.init_ckpt, map_locationdevice) model.load_state_dict(ckpt[model_state_dict]) print(fLoaded pretrained checkpoint from {args.init_ckpt}) optimizer torch.optim.Adam(model.parameters(), lrargs.lr) criterion nn.MSELoss() if args.mode pretrain: # 预训练只需要输入序列不需要标签 train_dataset TensorDataset(X_train_t) val_dataset TensorDataset(X_val_t) train_loader DataLoader(train_dataset, batch_sizeargs.batch_size, shuffleTrue) val_loader DataLoader(val_dataset, batch_sizeargs.batch_size, shuffleFalse) for epoch in range(args.epochs): model.train() train_loss 0.0 for (xb,) in train_loader: xb xb.to(device) masked_batch, mask mask_timesteps(xb, args.mask_ratio) pred model(masked_batch, modereconstruct) loss ((pred - xb) ** 2 * mask.float()).sum() / mask.float().sum() optimizer.zero_grad() loss.backward() optimizer.step() train_loss loss.item() * xb.size(0) # 验证 model.eval() val_loss 0.0 with torch.no_grad(): for (xb,) in val_loader: xb xb.to(device) masked_batch, mask mask_timesteps(xb, args.mask_ratio) pred model(masked_batch, modereconstruct) loss ((pred - xb) ** 2 * mask.float()).sum() / mask.float().sum() val_loss loss.item() * xb.size(0) train_loss / len(train_dataset) val_loss / len(val_dataset) print(fpretrain epoch {epoch 1}/{args.epochs}, loss: {train_loss:.4f}, val_loss: {val_loss:.4f}) else: # 微调预测下一时刻 train_dataset TensorDataset(X_train_t, Y_train_t) val_dataset TensorDataset(X_val_t, Y_val_t) train_loader DataLoader(train_dataset, batch_sizeargs.batch_size, shuffleTrue) val_loader DataLoader(val_dataset, batch_sizeargs.batch_size, shuffleFalse) for epoch in range(args.epochs): model.train() train_loss 0.0 for xb, yb in train_loader: xb xb.to(device) yb yb.to(device) pred model(xb, modeforecast) loss criterion(pred, yb) optimizer.zero_grad() loss.backward() optimizer.step() train_loss loss.item() * xb.size(0) val_loss, val_mae evaluate_forecast(model, val_loader, criterion, device) train_loss / len(train_dataset) print(ffinetune epoch {epoch 1}/{args.epochs}, loss: {train_loss:.4f}, val_loss: {val_loss:.4f}, val_mae: {val_mae:.4f}) # 5. 保存模型 torch.save({model_state_dict: model.state_dict()}, args.save_path) print(fModel saved to {args.save_path}) if __name__ __main__: main()6.2 预训练让模型学习空域通用规律运行下面的命令开始自监督预训练python train.py --mode pretrain --epochs 20 --save-path checkpoints/airspace_pretrain.pt预训练阶段的输入是连续的网格序列模型会随机把一部分时间步的网格值置为 0然后尝试还原。因为模型必须从其他时间步推测被遮挡的网格所以会学到空域流量在时间上的相关性比如热点区域流量如何变化、流量如何平滑转移等。预训练完成后checkpoints/airspace_pretrain.pt中保存的就是一个“见过大量空域运行规律”的通用模型。即使没有下游任务标签它也已经掌握了空域序列的基础表示。6.3 微调适配到流量预测任务使用预训练权重初始化模型然后在一小段时间序列上进行微调python train.py --mode finetune --epochs 30 --init-ckpt checkpoints/airspace_pretrain.pt --save-path checkpoints/airspace_finetune.pt微调阶段模型输入最近input_len个时间步的空域网格输出下一时刻的网格流量预测。损失函数使用 MSE验证集上会打印val_loss和val_mae。微调的作用是让模型从“还原序列”的能力转向“预测未来”的能力。预训练学到的时空表示会作为先验知识帮助模型在小样本条件下更快收敛这就是基础模型框架的核心价值。7. 推理与效果评估7.1 加载训练好的模型微调完成后可以用inference.py加载模型对最后一段历史数据执行预测。# inference.py import argparse import numpy as np import torch from data.synthetic_airspace import create_sequences, generate_airspace_data from model.airspace_transformer import AirspaceTransformer def normalize(data): min_val data.min() max_val data.max() return (data - min_val) / (max_val - min_val 1e-8) def main(): parser argparse.ArgumentParser() parser.add_argument(--ckpt, typestr, defaultcheckpoints/airspace_finetune.pt) parser.add_argument(--grid-size, typeint, default8) parser.add_argument(--d-model, typeint, default64) parser.add_argument(--nhead, typeint, default4) parser.add_argument(--num-layers, typeint, default2) parser.add_argument(--input-len, typeint, default24) args parser.parse_args() device torch.device(cuda if torch.cuda.is_available() else cpu) raw generate_airspace_data(num_timesteps500, grid_sizeargs.grid_size) data normalize(raw) X, Y create_sequences(data, input_lenargs.input_len, pred_len1) model AirspaceTransformer( grid_sizeargs.grid_size, d_modelargs.d_model, nheadargs.nhead, num_layersargs.num_layers, ).to(device) ckpt torch.load(args.ckpt, map_locationdevice) model.load_state_dict(ckpt[model_state_dict]) model.eval() # 取最后一个样本作为历史输入 history X[-1] # (input_len, grid_size, grid_size) true_next Y[-1][0] # (grid_size, grid_size) input_tensor torch.tensor(history, dtypetorch.float32).unsqueeze(0).to(device) with torch.no_grad(): pred model(input_tensor, modeforecast) pred pred.squeeze(0).cpu().numpy() print(Predicted next-step airspace grid:) print(np.round(pred, 3)) print(\nTrue next-step airspace grid:) print(np.round(true_next, 3)) print(\nAbsolute error:) print(np.round(np.abs(pred - true_next), 3)) if __name__ __main__: main()运行推理python inference.py --ckpt checkpoints/airspace_finetune.pt输出会打印预测网格、真实网格和逐格误差。你可以用 Matplotlib 把网格画成热力图更直观地观察预测区域和真实区域的差异。7.2 评估指标空域网格预测本质上是回归任务常用的指标有MSE均方误差对较大误差更敏感。MAE平均绝对误差更直观单位与网格流量一致。R2决定系数评估模型对目标方差解释的比例。如果预测值全部接近 0MAE 会很低但这时模型其实没有捕捉到空间结构。因此评估时不要只看单一指标还要结合热力图观察预测结果的空间形态是否合理。7.3 如何进一步提升效果在真实项目中可以尝试以下方法把输入序列拉长让模型看到更长时间范围的趋势。加入小时、星期等时间特征。加入机场容量、气象等外部特征。使用更大的d_model和更多 Encoder 层。使用多步预测的自回归评估观察误差累积情况。8. 常见问题与排查思路问题现象常见原因解决思路训练 loss 不下降学习率过大或过小数据未归一化检查数据范围调小或调大学习率微调后预测结果全为均值模型欠拟合序列长度不够增加层数或隐藏维度增大 input_len预训练 loss 很低但微调效果差预训练任务和下游任务差异大适当调整微调学习率增加微调 epoch显存不足网格过大、序列过长、batch 过大减小 grid_size、batch_size 或序列长度结果对热点区域不敏感合成数据热点分布单一模型容量不足增加热点数量扩展模型规模训练集和验证集结果差异大数据分布不一致按时间顺序划分避免随机打乱测试集加载模型时报参数不匹配预训练和微调模型结构不一致确保 grid_size、d_model 等参数一致排查时建议先做一个小规模实验固定随机种子用少量数据跑通流程再逐步扩大数据量和模型规模。这样能快速区分是代码问题还是模型效果问题。9. 工程落地与最佳实践9.1 数据合规与安全真实空域数据往往来自管制系统、地面雷达、ADS-B 地面站或航空公司运行系统。使用这些数据前必须确认数据使用授权、脱敏要求和安全边界。涉及位置和航班信息时要避免暴露敏感航班细节通常需要聚合到网格或扇区层面后再用于建模。9.2 数据质量重于模型结构基础模型效果的上限由数据质量决定。网格分辨率、时间粒度、缺失值处理都会直接影响预测效果。建议在建模前先做数据探索确认航班轨迹在网格中的分布是否均匀、是否大量存在缺失时段、是否存在系统偏差。9.3 从简单模型开始不要一开始就追求大模型。建议先用历史均值、线性回归、随机森林等简单方法建立基线再逐步引入 Transformer。如果简单模型已经达到业务要求就不需要为“基础模型”而使用基础模型。基础模型的价值主要体现在多场景迁移和复杂时空依赖建模上。9.4 训练与评估的切分方式时空序列数据不能随机划分训练集和验证集否则会造成时间泄漏。应按时间顺序切分比如用前 80% 的时间段训练后 20% 的时间段验证。如果业务存在明显周期性还要注意训练集中是否覆盖完整的周期。9.5 部署与监控模型上线后要监控输入数据分布是否发生漂移。例如空域结构调整、新航线开放、天气异常都会导致输入分布变化。建议定期用最近一段真实数据重新评估模型必要时做增量微调或重新训练。同时记录每次预测的时间、模型版本、输入数据版本方便回溯。9.6 性能与成本即使使用基础模型也不意味着一定要跑几百亿参数的模型。对于网格规模不大的空域场景几十层以内的 Transformer 已经能捕捉很多时空依赖。部署时可以考虑模型量化、批处理推理、缓存历史特征等方式降低延迟和计算成本。10. 下一步可以继续做什么本文从一个最小问题出发演示了“数据生成、预训练、微调、推理”的完整闭环。你可以在此基础上继续扩展使用真实 ADS-B 或航班计划数据替换合成数据但先确认数据使用授权。把网格结构换成图结构加入航路点、扇区边界等空域结构信息。在预训练阶段加入气象、机场容量等多模态输入提升模型的场景适应能力。尝试多步预测输出用自回归或直接序列生成方式预测未来 30/60 分钟的空域状态。增加数据规模测试模型容量的变化对预测效果的影响。空域预测是一个高度依赖领域知识的任务基础模型能带来泛化和迁移能力但数据、业务理解和工程落地同样重要。希望本文的代码和思路能给你一个清晰的起点帮助你在空域时空预测方向上走出第一步。 基于基础模型的空域预测：从预训练到微调的完整实现
近年来航班流量持续增长空域拥堵和航班延误已经成为民航运行中绕不开的难题。空管、机场、航司和飞行计划系统都有同一个需求提前知道未来一段时间某个空域会不会拥挤航班会怎么走流量会不会超过扇区负荷。这个需求背后就是“空域预测”。传统做法通常是为每个机场、每个扇区单独训练一个预测模型换一个场景就要重新标注、重新建模成本很高迁移效果不稳定。而“基础模型”的思路正好相反先在大规模、多来源的空域数据上进行预训练让模型学会空域运行的通用规律再通过少量业务数据微调适配到具体的流量预测、延误预测或扇区负荷预测任务上。本文将围绕“Foundation Model to Predict Airspace”这一主题从概念、数据构造、模型设计到预训练与微调完整演示一个最小可运行的空域预测系统。文章适合有 Python 基础、对深度学习时间序列建模感兴趣的开发者阅读。读完你不仅能理解基础模型在空域场景的应用思路还能拿到一套可以本地运行的代码示例并了解真实落地时需要注意的工程问题。1. 空域预测为什么需要基础模型1.1 空域预测到底在预测什么空域是一个连续的三维空间但实际业务中往往不会直接对连续空间建模而是把某一高度层范围内的空域切成网格每个网格单元记录一段时间内的航班数量、平均飞行速度、爬升下降状态等。这样空域预测就变成了“网格时空序列”预测问题。常见任务包括网格流量预测预测下一个时间片每个网格单元内的航班数量。扇区负荷预测预测某个管制扇区未来 15/30/60 分钟的交通负荷。航迹补全与预测根据历史轨迹预测飞机未来位置。延误传播预测某机场延误后影响如何扩散到周边空域。这些任务虽然名称不同但底层的时空特征高度相似都依赖航班位置、速度、气象、机场容量等信息都存在明显的时间和空间关联。这正是基础模型可以发挥价值的地方。1.2 什么是基础模型基础模型Foundation Model是指在大规模数据上进行预训练学习到通用领域知识再通过微调适配到多个下游任务的模型。它和传统“一个任务一个模型”的最大区别在于“通用性”。在文本领域BERT、GPT 都是典型的基础模型先在大规模语料上学习语言规律再用于分类、生成、问答等任务。在时空预测领域基础模型的思路同样成立先让模型在大规模空域运行数据上学习时空变化的通用模式比如高峰小时的流量形态、天气影响下的流量下降、机场放行率对区域流量的传导等再通过少量目标场景数据微调得到业务可用的预测模型。1.3 基础模型相对传统方案的优势传统流量预测常用 ARIMA、LSTM、CNN 或 GBDT这些模型在小规模单场景数据上可以做得不错但有几个明显限制场景迁移能力弱A 机场训练的模型很难直接用于 B 机场。特征工程成本高每个场景都要人工选择特征。长距离时空依赖建模困难传统序列模型对远距离空间交互和长周期时间依赖的捕捉有限。基础模型通过预训练和微调把大部分通用规律沉淀在模型参数中。新场景只需要少量数据即可适配同时 Transformer 结构天然适合建模长距离依赖也方便融入网格、轨迹、气象等多模态信息。1.4 基础模型在空域预测中的落地路径在空域场景中基础模型并不是“拿来即用”的通用大模型而是需要结合领域数据的“预训练微调”框架数据准备收集历史航班轨迹、航班计划、气象数据、机场运行数据。网格化处理把原始航班数据映射到统一时空网格。自监督预训练用大量未标注数据训练模型还原被遮挡的时空片段学习空域运行的通用规律。下游微调用少量标注数据训练模型预测网格流量、扇区负荷或延误。推理与部署将模型接入运行系统输出未来时刻的空域状态。接下来的章节我们就按这个路径用一份合成数据搭建一个最小可运行的演示系统。2. 技术路线从时空网格到预训练-微调2.1 数据层空域预测可用的数据源很多常见包括ADS-B / 雷达轨迹提供航班经度、纬度、高度、速度、航向等实时信息。航班计划起飞时间、降落时间、航路、机型。气象数据风场、能见度、雷暴区域。机场运行数据跑道容量、放行率、延误状态。空域结构数据扇区边界、航路点、限制区。真实项目中数据质量是最大的挑战。本文为了方便演示使用合成数据模拟航班流量在空域网格上的分布和变化。2.2 表示层把原始空域数据组织成模型能理解的张量常见有两种思路时空网格序列把空域按经纬度切成H x W的网格每个时间片生成一个(H, W)矩阵连续时间片组成(T, H, W)序列。图结构序列把航路点和扇区看作节点航路看作边用图神经网络建模节点间的流量传播。本文采用第一种方式优点是实现简单、通用性强。真实场景如果航路结构明显可以在此基础上扩展为图结构。2.3 模型层模型采用基于 Transformer 的编码器核心组件包括输入嵌入层把(H, W)网格展平成向量映射到d_model维。位置编码层给每个时间步加入时序位置信息。Transformer Encoder捕捉时间维度的依赖关系。输出投影层根据任务输出重构序列或预测未来网格。整个训练过程分为两个阶段预训练阶段随机遮挡输入序列中的部分时间步模型需要还原被遮挡的网格值这就是自监督学习。微调阶段使用未来时刻网格值作为标签让模型输出下一时刻的预测值。整体流程可以简化为历史网格序列 (T, H, W) | v 随机遮挡 - 预训练还原 - 得到通用空域模型 | v 输入最近 T 个时间片 - Transformer - 下一时刻网格预测3. 环境准备与项目结构3.1 运行环境示例代码基于 Python 和 PyTorch依赖较少适合本地学习。建议环境如下Python 3.9 或以上版本PyTorch 2.0 或以上版本NumPyPandas可选用于数据处理Matplotlib可选用于结果可视化scikit-learn可选用于指标计算安装依赖可以使用以下命令pip install torch numpy pandas matplotlib scikit-learn本示例没有使用 CUDA 也能运行只是训练速度会慢一些。如果机器有 GPUPyTorch 会自动使用 GPU 加速。3.2 项目结构为了保持代码清晰建议按下面的目录组织项目airspace_foundation/ ├── data/ │ └── synthetic_airspace.py ├── model/ │ └── airspace_transformer.py ├── train.py ├── inference.py └── checkpoints/ # 训练后保存模型权重其中data目录负责生成合成空域数据model目录定义基础模型结构train.py完成预训练和微调inference.py加载训练好的模型进行预测。4. 构造空域时序数据集4.1 为什么使用合成数据真实空域数据涉及空管系统数据使用协议、航班隐私、安全边界等问题并不是所有人都能方便获取。为了演示完整的建模流程我们先用合成数据验证方法和代码。合成数据的设计要尽量贴近真实空域的流量特点有高峰和低谷、有局部热点区域、有时间变化趋势。4.2 合成数据生成逻辑下面代码会生成一个[num_timesteps, grid_size, grid_size]的三维数组表示空域网格每个时间片的航班流密度。我们随机设置了几个热点中心每个中心附近流量较高整体流量随时间呈正弦波动。# data/synthetic_airspace.py import numpy as np def generate_airspace_data(num_timesteps1000, grid_size8, num_centers3, seed42): 生成合成的空域网格流量数据。 参数 num_timesteps: 生成的时间步数量 grid_size: 空域网格的边长 num_centers: 热点中心数量 seed: 随机种子 返回 shape 为 (num_timesteps, grid_size, grid_size) 的 float32 数组 rng np.random.default_rng(seed) # 随机生成热点中心坐标 centers rng.uniform(0, grid_size, size(num_centers, 2)) strengths rng.uniform(0.8, 2.0, sizenum_centers) data np.zeros((num_timesteps, grid_size, grid_size), dtypenp.float32) for t in range(num_timesteps): field np.full((grid_size, grid_size), 0.5, dtypenp.float32) for i in range(grid_size): for j in range(grid_size): # 取网格中心点坐标方便计算距离 x, y i 0.5, j 0.5 for (cx, cy), strength in zip(centers, strengths): dist np.sqrt((x - cx) ** 2 (y - cy) ** 2) # 热点中心附近流量更高并叠加时间变化 field[i, j] strength * np.exp(-dist) * (1 0.3 * np.sin(t / 20)) data[t] field return data def create_sequences(data, input_len24, pred_len1): 把原始网格数据切分为输入序列和预测目标。 参数 data: shape 为 (num_timesteps, grid_size, grid_size) input_len: 输入的时间步数 pred_len: 预测的未来时间步数 返回 X: shape 为 (样本数, input_len, grid_size, grid_size) Y: shape 为 (样本数, pred_len, grid_size, grid_size) X, Y [], [] for t in range(len(data) - input_len - pred_len 1): X.append(data[t:t input_len]) Y.append(data[t input_len:t input_len pred_len]) return np.stack(X), np.stack(Y)核心逻辑说明热点中心用均匀分布随机生成模拟不同空域结构下的流量密集区域。strength * np.exp(-dist)模拟距离热点越远流量越低的空间衰减。np.sin(t / 20)模拟一天或一段时间内的流量波动。create_sequences使用滑动窗口把连续时间片切成训练样本。4.3 数据归一化训练神经网络前需要对数据进行归一化否则数值范围差距过大会影响收敛。下面是一个简单的 Min-Max 归一化函数def normalize(data): 将数据缩放到 [0, 1] 区间。 min_val data.min() max_val data.max() return (data - min_val) / (max_val - min_val 1e-8)注意在真实项目中归一化参数只能在训练集上计算不能使用验证集和测试集的信息否则会造成数据泄漏。5. 实现基础模型AirspaceTransformer5.1 位置编码Transformer 本身不包含时序位置信息因此需要显式加入位置编码。常用的正弦位置编码实现如下# model/airspace_transformer.py import math import torch import torch.nn as nn class PositionalEncoding(nn.Module): 标准正弦位置编码。 def __init__(self, d_model, max_len512): super().__init__() pe torch.zeros(max_len, d_model) position torch.arange(0, max_len).unsqueeze(1).float() div_term torch.exp( torch.arange(0, d_model, 2).float() * (-math.log(10000.0) / d_model) ) pe[:, 0::2] torch.sin(position * div_term) pe[:, 1::2] torch.cos(position * div_term) self.register_buffer(pe, pe) def forward(self, x): # x: (batch, seq_len, d_model) return x self.pe[: x.size(1)]这里的register_buffer会把位置编码注册为模型参数的一部分但它不需要更新梯度模型保存和加载时会一起保存。5.2 AirspaceTransformer 模型模型的核心结构是 Transformer Encoder同时支持两种输出模式reconstruct模式输出整个输入序列每个时间步的网格值用于预训练。forecast模式只输出最后一个时间步之后的一帧预测结果用于下游微调。class AirspaceTransformer(nn.Module): 面向空域网格序列的基础模型。 参数 grid_size: 空域网格边长 d_model: Transformer 隐藏维度 nhead: 多头注意力头数 num_layers: Transformer Encoder 层数 max_len: 位置编码最大长度 def __init__(self, grid_size8, d_model64, nhead4, num_layers2, max_len512): super().__init__() self.grid_size grid_size self.d_model d_model # 输入嵌入层把每个时间步的网格展平并映射到 d_model self.input_embedding nn.Linear(grid_size * grid_size, d_model) # 位置编码 self.pos_encoder PositionalEncoding(d_model, max_len) # Transformer Encoder encoder_layer nn.TransformerEncoderLayer( d_modeld_model, nheadnhead, batch_firstTrue ) self.encoder nn.TransformerEncoder(encoder_layer, num_layersnum_layers) # 输出投影层映射回网格维度 self.output_proj nn.Linear(d_model, grid_size * grid_size) def forward(self, x, modeforecast): 参数 x: (batch, seq_len, grid_size, grid_size) mode: reconstruct 表示还原全序列 forecast 表示预测下一时刻 batch, seq_len, h, w x.shape # 把网格展平成特征向量 x x.reshape(batch, seq_len, h * w) x self.input_embedding(x) x self.pos_encoder(x) x self.encoder(x) if mode reconstruct: # 还原每个时间步的网格 out self.output_proj(x) return out.reshape(batch, seq_len, h, w) elif mode forecast: # 使用最后一个时间步的编码结果预测下一时刻 last x[:, -1, :] out self.output_proj(last) return out.reshape(batch, h, w) else: raise ValueError(fUnsupported mode: {mode})这个模型虽然规模不大但已经具备基础模型“共享参数、多任务输出”的特征。预训练阶段用reconstruct模式学习通用空域规律微调阶段用forecast模式输出下一时刻预测。5.3 为什么选择这样的模型结构选择 Transformer Encoder 而不是更复杂的 Encoder-Decoder是因为本文演示的任务是“根据历史还原未来”本质上更接近序列表示学习。Encoder 可以直接输出每个时间步的隐含表示便于同时支持重构和预测。真实项目中如果要做多步预测或生成完整轨迹可以在这个基础上扩展 Decoder 结构。6. 预训练与下游微调完整实战6.1 训练脚本总体设计train.py是整个训练流程的入口支持两个模式pretrain自监督预训练随机遮挡部分时间步让模型还原遮挡位置。finetune使用历史序列预测下一时刻加载预训练权重后微调。训练脚本中加入了随机种子控制保证实验结果可复现。# train.py import argparse import numpy as np import torch import torch.nn as nn from torch.utils.data import DataLoader, TensorDataset from data.synthetic_airspace import create_sequences, generate_airspace_data from model.airspace_transformer import AirspaceTransformer def normalize(data): min_val data.min() max_val data.max() return (data - min_val) / (max_val - min_val 1e-8) def mask_timesteps(batch, mask_ratio0.15): 随机遮挡部分时间步用于自监督预训练。 参数 batch: (batch_size, seq_len, grid_size, grid_size) mask_ratio: 遮挡比例 返回 masked_batch: 遮挡后的输入 mask: 遮挡掩码True 表示该位置被遮挡 mask torch.rand(batch.size(0), batch.size(1), 1, 1, devicebatch.device) mask mask mask_ratio mask mask.expand_as(batch) masked_batch batch.masked_fill(mask, 0.0) return masked_batch, mask def evaluate_forecast(model, loader, criterion, device): 评估预测模式下的损失和平均绝对误差。 model.eval() total_loss 0.0 total_mae 0.0 with torch.no_grad(): for xb, yb in loader: xb xb.to(device) yb yb.to(device) pred model(xb, modeforecast) total_loss criterion(pred, yb).item() * xb.size(0) total_mae (pred - yb).abs().sum().item() n len(loader.dataset) return total_loss / n, total_mae / n def main(): parser argparse.ArgumentParser(descriptionAirspace Foundation Model Training) parser.add_argument(--mode, typestr, choices[pretrain, finetune], requiredTrue) parser.add_argument(--num-timesteps, typeint, default1200) parser.add_argument(--grid-size, typeint, default8) parser.add_argument(--input-len, typeint, default24) parser.add_argument(--d-model, typeint, default64) parser.add_argument(--nhead, typeint, default4) parser.add_argument(--num-layers, typeint, default2) parser.add_argument(--batch-size, typeint, default32) parser.add_argument(--epochs, typeint, default20) parser.add_argument(--lr, typefloat, default1e-3) parser.add_argument(--mask-ratio, typefloat, default0.15) parser.add_argument(--init-ckpt, typestr, defaultNone, help微调时加载预训练模型) parser.add_argument(--save-path, typestr, defaultcheckpoints/airspace_model.pt) args parser.parse_args() torch.manual_seed(42) np.random.seed(42) device torch.device(cuda if torch.cuda.is_available() else cpu) print(fUsing device: {device}) # 1. 生成并归一化数据 raw generate_airspace_data( num_timestepsargs.num_timesteps, grid_sizeargs.grid_size, ) data normalize(raw) # 2. 划分序列数据集 X, Y create_sequences(data, input_lenargs.input_len, pred_len1) split_idx int(len(X) * 0.8) X_train, X_val X[:split_idx], X[split_idx:] Y_train, Y_val Y[:split_idx], Y[split_idx:] # 3. 包装成 PyTorch Dataset X_train_t torch.tensor(X_train, dtypetorch.float32) X_val_t torch.tensor(X_val, dtypetorch.float32) Y_train_t torch.tensor(Y_train, dtypetorch.float32).squeeze(1) Y_val_t torch.tensor(Y_val, dtypetorch.float32).squeeze(1) # 4. 初始化模型 model AirspaceTransformer( grid_sizeargs.grid_size, d_modelargs.d_model, nheadargs.nhead, num_layersargs.num_layers, ).to(device) if args.mode finetune and args.init_ckpt is not None: ckpt torch.load(args.init_ckpt, map_locationdevice) model.load_state_dict(ckpt[model_state_dict]) print(fLoaded pretrained checkpoint from {args.init_ckpt}) optimizer torch.optim.Adam(model.parameters(), lrargs.lr) criterion nn.MSELoss() if args.mode pretrain: # 预训练只需要输入序列不需要标签 train_dataset TensorDataset(X_train_t) val_dataset TensorDataset(X_val_t) train_loader DataLoader(train_dataset, batch_sizeargs.batch_size, shuffleTrue) val_loader DataLoader(val_dataset, batch_sizeargs.batch_size, shuffleFalse) for epoch in range(args.epochs): model.train() train_loss 0.0 for (xb,) in train_loader: xb xb.to(device) masked_batch, mask mask_timesteps(xb, args.mask_ratio) pred model(masked_batch, modereconstruct) loss ((pred - xb) ** 2 * mask.float()).sum() / mask.float().sum() optimizer.zero_grad() loss.backward() optimizer.step() train_loss loss.item() * xb.size(0) # 验证 model.eval() val_loss 0.0 with torch.no_grad(): for (xb,) in val_loader: xb xb.to(device) masked_batch, mask mask_timesteps(xb, args.mask_ratio) pred model(masked_batch, modereconstruct) loss ((pred - xb) ** 2 * mask.float()).sum() / mask.float().sum() val_loss loss.item() * xb.size(0) train_loss / len(train_dataset) val_loss / len(val_dataset) print(fpretrain epoch {epoch 1}/{args.epochs}, loss: {train_loss:.4f}, val_loss: {val_loss:.4f}) else: # 微调预测下一时刻 train_dataset TensorDataset(X_train_t, Y_train_t) val_dataset TensorDataset(X_val_t, Y_val_t) train_loader DataLoader(train_dataset, batch_sizeargs.batch_size, shuffleTrue) val_loader DataLoader(val_dataset, batch_sizeargs.batch_size, shuffleFalse) for epoch in range(args.epochs): model.train() train_loss 0.0 for xb, yb in train_loader: xb xb.to(device) yb yb.to(device) pred model(xb, modeforecast) loss criterion(pred, yb) optimizer.zero_grad() loss.backward() optimizer.step() train_loss loss.item() * xb.size(0) val_loss, val_mae evaluate_forecast(model, val_loader, criterion, device) train_loss / len(train_dataset) print(ffinetune epoch {epoch 1}/{args.epochs}, loss: {train_loss:.4f}, val_loss: {val_loss:.4f}, val_mae: {val_mae:.4f}) # 5. 保存模型 torch.save({model_state_dict: model.state_dict()}, args.save_path) print(fModel saved to {args.save_path}) if __name__ __main__: main()6.2 预训练让模型学习空域通用规律运行下面的命令开始自监督预训练python train.py --mode pretrain --epochs 20 --save-path checkpoints/airspace_pretrain.pt预训练阶段的输入是连续的网格序列模型会随机把一部分时间步的网格值置为 0然后尝试还原。因为模型必须从其他时间步推测被遮挡的网格所以会学到空域流量在时间上的相关性比如热点区域流量如何变化、流量如何平滑转移等。预训练完成后checkpoints/airspace_pretrain.pt中保存的就是一个“见过大量空域运行规律”的通用模型。即使没有下游任务标签它也已经掌握了空域序列的基础表示。6.3 微调适配到流量预测任务使用预训练权重初始化模型然后在一小段时间序列上进行微调python train.py --mode finetune --epochs 30 --init-ckpt checkpoints/airspace_pretrain.pt --save-path checkpoints/airspace_finetune.pt微调阶段模型输入最近input_len个时间步的空域网格输出下一时刻的网格流量预测。损失函数使用 MSE验证集上会打印val_loss和val_mae。微调的作用是让模型从“还原序列”的能力转向“预测未来”的能力。预训练学到的时空表示会作为先验知识帮助模型在小样本条件下更快收敛这就是基础模型框架的核心价值。7. 推理与效果评估7.1 加载训练好的模型微调完成后可以用inference.py加载模型对最后一段历史数据执行预测。# inference.py import argparse import numpy as np import torch from data.synthetic_airspace import create_sequences, generate_airspace_data from model.airspace_transformer import AirspaceTransformer def normalize(data): min_val data.min() max_val data.max() return (data - min_val) / (max_val - min_val 1e-8) def main(): parser argparse.ArgumentParser() parser.add_argument(--ckpt, typestr, defaultcheckpoints/airspace_finetune.pt) parser.add_argument(--grid-size, typeint, default8) parser.add_argument(--d-model, typeint, default64) parser.add_argument(--nhead, typeint, default4) parser.add_argument(--num-layers, typeint, default2) parser.add_argument(--input-len, typeint, default24) args parser.parse_args() device torch.device(cuda if torch.cuda.is_available() else cpu) raw generate_airspace_data(num_timesteps500, grid_sizeargs.grid_size) data normalize(raw) X, Y create_sequences(data, input_lenargs.input_len, pred_len1) model AirspaceTransformer( grid_sizeargs.grid_size, d_modelargs.d_model, nheadargs.nhead, num_layersargs.num_layers, ).to(device) ckpt torch.load(args.ckpt, map_locationdevice) model.load_state_dict(ckpt[model_state_dict]) model.eval() # 取最后一个样本作为历史输入 history X[-1] # (input_len, grid_size, grid_size) true_next Y[-1][0] # (grid_size, grid_size) input_tensor torch.tensor(history, dtypetorch.float32).unsqueeze(0).to(device) with torch.no_grad(): pred model(input_tensor, modeforecast) pred pred.squeeze(0).cpu().numpy() print(Predicted next-step airspace grid:) print(np.round(pred, 3)) print(\nTrue next-step airspace grid:) print(np.round(true_next, 3)) print(\nAbsolute error:) print(np.round(np.abs(pred - true_next), 3)) if __name__ __main__: main()运行推理python inference.py --ckpt checkpoints/airspace_finetune.pt输出会打印预测网格、真实网格和逐格误差。你可以用 Matplotlib 把网格画成热力图更直观地观察预测区域和真实区域的差异。7.2 评估指标空域网格预测本质上是回归任务常用的指标有MSE均方误差对较大误差更敏感。MAE平均绝对误差更直观单位与网格流量一致。R2决定系数评估模型对目标方差解释的比例。如果预测值全部接近 0MAE 会很低但这时模型其实没有捕捉到空间结构。因此评估时不要只看单一指标还要结合热力图观察预测结果的空间形态是否合理。7.3 如何进一步提升效果在真实项目中可以尝试以下方法把输入序列拉长让模型看到更长时间范围的趋势。加入小时、星期等时间特征。加入机场容量、气象等外部特征。使用更大的d_model和更多 Encoder 层。使用多步预测的自回归评估观察误差累积情况。8. 常见问题与排查思路问题现象常见原因解决思路训练 loss 不下降学习率过大或过小数据未归一化检查数据范围调小或调大学习率微调后预测结果全为均值模型欠拟合序列长度不够增加层数或隐藏维度增大 input_len预训练 loss 很低但微调效果差预训练任务和下游任务差异大适当调整微调学习率增加微调 epoch显存不足网格过大、序列过长、batch 过大减小 grid_size、batch_size 或序列长度结果对热点区域不敏感合成数据热点分布单一模型容量不足增加热点数量扩展模型规模训练集和验证集结果差异大数据分布不一致按时间顺序划分避免随机打乱测试集加载模型时报参数不匹配预训练和微调模型结构不一致确保 grid_size、d_model 等参数一致排查时建议先做一个小规模实验固定随机种子用少量数据跑通流程再逐步扩大数据量和模型规模。这样能快速区分是代码问题还是模型效果问题。9. 工程落地与最佳实践9.1 数据合规与安全真实空域数据往往来自管制系统、地面雷达、ADS-B 地面站或航空公司运行系统。使用这些数据前必须确认数据使用授权、脱敏要求和安全边界。涉及位置和航班信息时要避免暴露敏感航班细节通常需要聚合到网格或扇区层面后再用于建模。9.2 数据质量重于模型结构基础模型效果的上限由数据质量决定。网格分辨率、时间粒度、缺失值处理都会直接影响预测效果。建议在建模前先做数据探索确认航班轨迹在网格中的分布是否均匀、是否大量存在缺失时段、是否存在系统偏差。9.3 从简单模型开始不要一开始就追求大模型。建议先用历史均值、线性回归、随机森林等简单方法建立基线再逐步引入 Transformer。如果简单模型已经达到业务要求就不需要为“基础模型”而使用基础模型。基础模型的价值主要体现在多场景迁移和复杂时空依赖建模上。9.4 训练与评估的切分方式时空序列数据不能随机划分训练集和验证集否则会造成时间泄漏。应按时间顺序切分比如用前 80% 的时间段训练后 20% 的时间段验证。如果业务存在明显周期性还要注意训练集中是否覆盖完整的周期。9.5 部署与监控模型上线后要监控输入数据分布是否发生漂移。例如空域结构调整、新航线开放、天气异常都会导致输入分布变化。建议定期用最近一段真实数据重新评估模型必要时做增量微调或重新训练。同时记录每次预测的时间、模型版本、输入数据版本方便回溯。9.6 性能与成本即使使用基础模型也不意味着一定要跑几百亿参数的模型。对于网格规模不大的空域场景几十层以内的 Transformer 已经能捕捉很多时空依赖。部署时可以考虑模型量化、批处理推理、缓存历史特征等方式降低延迟和计算成本。10. 下一步可以继续做什么本文从一个最小问题出发演示了“数据生成、预训练、微调、推理”的完整闭环。你可以在此基础上继续扩展使用真实 ADS-B 或航班计划数据替换合成数据但先确认数据使用授权。把网格结构换成图结构加入航路点、扇区边界等空域结构信息。在预训练阶段加入气象、机场容量等多模态输入提升模型的场景适应能力。尝试多步预测输出用自回归或直接序列生成方式预测未来 30/60 分钟的空域状态。增加数据规模测试模型容量的变化对预测效果的影响。空域预测是一个高度依赖领域知识的任务基础模型能带来泛化和迁移能力但数据、业务理解和工程落地同样重要。希望本文的代码和思路能给你一个清晰的起点帮助你在空域时空预测方向上走出第一步。