简介：基于STK11场景的卫星任务调度与强化学习训练数据生成系统，面向卫星任务规划与机器学习交叉领域的研究者或工程师，提供从随机观测任务生成、卫星可访问时段计算、数据对齐与批次排序，到数据增强、模型训练及奖励可视化的完整链路解决方案。压缩包共包含205个文件，约79.58MB，主要有13个Python处理脚本、39个CSV训练数据、63张PNG可视化图表、7个PTH模型权重，以及STK场景文件（SA、SN、SC等）和DOCX/TXT说明文档，方便直接对照工程结构和说明进行复现。目前已有67人学习下载，适合需要构建大规模卫星观测任务数据集的读者参考。资源内随机任务生成、访问时段计算、数据对齐、批次排序、数据增强等模块均有对应代码与样例数据，配合模型训练和奖励可视化结果，能够帮助快速理解强化学习在卫星调度中的应用流程；相关脚本和工程结构可直接修改复用，服务于航天、地理信息、环境监测等方向的算法验证与仿真应用。
1. 基于STK11场景的卫星任务调度强化学习数据生成：卡点从来不在算法
基于STK11场景做卫星任务调度与强化学习训练数据生成，最反直觉的一点是训练不卡算法，卡在数据。公开的卫星调度数据集要么只有几十个场景，要么把访问时段提前算好、丢给你一个抽象矩阵，换个星座布局立刻失效。这套系统的思路是把STK11当几何真值源：STK负责算卫星对地面目标的访问时段，外部脚本负责随机任务生成，再把访问时段和任务窗口做数据对齐处理，经过批次排序与数据增强后进入模型训练，最后用奖励可视化确认策略学到了调度规律。适合在轨任务调度、遥感星座运营，以及想从运筹优化转强化学习的工程师。
2. STK11场景建模与随机任务生成：先建沙盘，再撒任务
整套系统按三层来拆最容易落地：STK11提供场景与轨道几何，脚本层负责随机任务生成和落盘，训练层消费前两层产出的数据。常见做法不是手工在GUI里逐颗卫星点，而是通过STK11的COM接口把场景创建、轨道设置、目标增删全部参数化，这样换一套星座布局只是换一个配置文件。
2.1 用Windows COM接口把STK11变成可编程调度沙盘
STK11在Windows下暴露了Application对象，Python用pywin32就能直接驱动。先拿Application，再取Personality2根对象，后续的对象创建、命令执行都挂在root上。
import win32com.client from win32com.client import constants as C stk = win32com.client.Dispatch("STK11.Application") stk.Visible = True stk.UserControl = False root = stk.Personality2 root.NewScenario("obs_gen") sc = root.CurrentScenario sc.SetTimePeriod("1 Jan 2025 00:00:00.00", "1 Jan 2025 06:00:00.00") sat = sc.Children.New(C.eSatellite, "Sat-0") sat.SetPropagatorType(C.ePropagatorJ4Perturbation) root.ExecuteCommand( 'SetState */Satellite/Sat-0 Classical J4Perturbation ' '"1 Jan 2025 00:00:00.00" J2000 ' '6878.0 0.00124 51.6 30.0 0.0 0.0 ' '60 "1 Jan 2025 06:00:00.00"' )说明：场景设6小时是权衡结果。太短单星对同一目标的访问次数太少，训练样本里“二次访问”这类关键状态根本不会出现；太长J4传播步数多，批量化生成脚本慢到没法迭代。SetState里J2000之后依次是半长轴(km)、偏心率、倾角、升交点赤经、近地点幅角、平近点角，然后是积分步长和结束时刻。这条命令在不同补丁版本里对关键字大小写有差异，批量跑前先拿单星试一条，确认返回success再放开循环。
提示：不要在一开始就上HPOP。调度训练数据要的是访问几何而不是精密定轨，HPOP考虑阻力、光压和第三方引力，一颗星6小时弧段要算几十秒，J4只要秒级。用来生成训练样本，两者访问窗口差异通常在10秒以内，远小于后面要引入的任务时间窗抖动范围。
批量化建模用配置文件控制星座布局
我一般把半长轴6800~7200km、倾角51.6°和太阳同步两种倾向按比例混合，升交点赤经均匀散布。这样生成的访问时段能覆盖不同纬度带，训练出来的策略不会只认识一个星座。目标用PointTarget而不是AreaTarget，点目标参与访问计算比区域边界快一个量级；等最终验证集再换AreaTarget重算一批做比对。
2.2 随机任务生成：四个参数组而不是随机撒点
随机任务如果只随机经纬度，生成的数据没有调度难度梯度，策略会停留在“见窗就接”的水平。我一般把生成逻辑拆成四个参数组：
| 参数组 | 字段 | 典型取值 | 作用 |
|---|---|---|---|
| 空间分布 | lon/lat范围、区域密度 | 全球或按区域加权 | 决定星座几何覆盖难易 |
| 时间分布 | 释放时间、截止时间、窗口长度 | 900/1800/3600 s | 决定时间竞争强度 |
| 服务需求 | 观测时长、任务类型 | 成像/数传/应急，5~40 s | 决定单星可串行的任务数 |
| 优先级 | priority、惩罚权重 | 1/2/3按0.5/0.35/0.15 | 决定奖励函数的相对大小 |
import numpy as np import pandas as pd def gen_tasks(n, epoch_sec, rng): """生成n个原始任务：只写属性，不做任何几何计算""" rows = [] for i in range(n): lon = rng.uniform(-180.0, 180.0) lat = rng.uniform(-55.0, 55.0) tw_len = float(rng.choice([900, 1800, 3600])) tw_start = rng.uniform(0.0, epoch_sec - tw_len) # 防窗口越界 dur = rng.uniform(5.0, 40.0) prio = int(rng.choice([1, 2, 3], p=[0.50, 0.35, 0.15])) ttype = rng.choice(["IMG", "COMM", "EMER"], p=[0.6, 0.3, 0.1]) rows.append(dict(task_id=i, lon=lon, lat=lat, tw_start=tw_start, tw_end=tw_start + tw_len, dur=dur, prio=prio, type=ttype)) return pd.DataFrame(rows)说明：tw_start必须在[0, epoch_sec - tw_len]里取，否则一批任务全挤在场景末尾，可访问时段样本几乎为零。EMER对应应急观测，窗口给得更短，用来模拟高时间竞争。窗口长度、时长、优先级三个分布直接决定数据集难度，生成完先画三个字段的直方图再决定要不要调参。
2.3 落盘格式：用Parquet组织大规模观测任务数据
任务池和访问记录要跨训练循环反复读取，这套系统的目标是“大规模卫星观测任务数据”，动不动几百个episode、上千万行访问记录，CSV的解析开销吃不消。我按episode一个Parquet文件落盘，时间统一转成相对场景起点的浮点秒。
import pyarrow as pa import pyarrow.parquet as pq schema = pa.schema([ ("episode_id", pa.int32()), ("task_id", pa.int32()), ("sat_id", pa.int32()), ("acc_start", pa.float64()), ("acc_end", pa.float64()), ("tw_start", pa.float64()), ("tw_end", pa.float64()), ("dur", pa.float64()), ("prio", pa.int32()), ]) tab = pa.Table.from_pandas(access_df[schema.names], schema=schema) pq.write_table(tab, f"ep_{ep_id:04d}.parquet", compression="zstd")说明：文件名里ep是episode编号，训练脚本按编号读取即可复现。zstd压缩对浮点时间列压缩比一般在3~5倍。每个episode目录里再放一个meta.json，记录生成种子、轨道参数、增强参数和丢弃率，训练脚本启动时读meta并写进TensorBoard的hparams，回头对比实验不用翻git记录。
3. 卫星可访问时段计算与数据对齐处理：从几何可见到调度可用
STK算出来的“可见”和调度能用的“可观测”是两码事。几何可见只代表传感器能看到目标，调度层还要满足任务窗口、最短观测时长、卫星资源余量三个条件。这一章就是把STK的访问计算结果加工成RL能读的状态矩阵。
3.1 访问计算的批量组织：先粗过滤再做STK Access
卫星目标做全排列访问计算，N颗星乘M个目标就是N×M次ComputeAccess。假设6颗星对1000个目标，6000次计算每轮场景都跑，脚本会慢到没法迭代。我一般先做纬度带粗过滤，再调用STK Access。
def compute_access_batch(root, sats, targets): """按纬度带粗过滤后再算Access，减少无效STK调用""" results = [] for sat in sats: for tgt in targets: if abs(sat.subpoint_lat - tgt.lat) > 30.0: continue acc = sat.GetAccessToObject(tgt) acc.ComputeAccess() dp = acc.DataProviders.Item("Access Data") res = dp.Exec(root.CurrentScenario.StartTime, root.CurrentScenario.StopTime, "") ds = res.DataSets[0] rows = list(zip(ds.GetValues()["Start Time"], ds.GetValues()["End Time"])) results.extend((sat.id, tgt.id, s, e) for s, e in rows) return results说明：subpoint_lat是上一轮传播后缓存的最大星下点纬度，属于近似值，只用来滤掉明显没有交集的星地配对；阈值宁可设大不要设小，漏掉真实访问段影响的是样本覆盖。真正口径以STK的Access结果为准，粗过滤只影响运行速度不影响精度。
3.2 时间窗对齐：重叠区间与最短观测时长的判定
拿到访问段后和任务窗口求交集，这就是整套系统里“数据对齐处理”的核心。两个时间窗的重叠如果覆盖不了观测时长，这个样本即使写进数据集，训练时也永远学不到正奖励。
def align_access(acc_df, task_df): """对齐访问时段与任务窗口，只保留能满足观测时长的记录""" merged = acc_df.merge(task_df, on="target_id") ov_s = merged[["acc_start", "tw_start"]].max(axis=1) ov_e = merged[["acc_end", "tw_end"]].min(axis=1) merged["ov_s"] = ov_s merged["ov_e"] = ov_e merged["feasible"] = (ov_e - ov_s) >= merged["dur"] return merged[merged["feasible"]].copy()说明：ov_s取两个时间窗各自开始时刻的较大值，ov_e取较小值，两者之差就是几何重叠区间。feasible列是后面模型训练里action mask的依据，feasible=False的候选在状态里直接置0，不允许智能体选择。对齐处理最容易踩的坑是时间单位不一致，STK报告能导出UTC字符串也能导出距起点秒数，混用会让重叠区间凭空多几小时，生成阶段统一用浮点秒，落盘前用pandas校验数据类型。
注意：merge之前不要删掉没有访问记录的任务id。全程不可见的任务要保留成负样本，这类样本对训练“拒绝”动作至关重要。
3.3 状态特征对齐：把轨道几何转成强化学习观测向量
访问时段对齐解决“能不能接”，状态特征对齐解决“怎么描述当前局面”。卫星ECI坐标是几千公里量级，任务窗口是秒级，直接拼在一起网络很难收敛。我做两类处理：时间特征相对化，几何特征归一化。
def build_obs(feat, ep_start, dims): """把一帧调度状态拼成定长向量；dims保存归一化上下界""" obs = np.zeros(dims["total"], dtype=np.float32) obs[0] = np.sin(2 * np.pi * (feat["t"] - ep_start) / 86400.0) obs[1] = np.cos(2 * np.pi * (feat["t"] - ep_start) / 86400.0) obs[2:5] = ((feat["sat_pos_eci"] - dims["pos_lo"]) / (dims["pos_hi"] - dims["pos_lo"])) obs[5] = feat["energy"] / feat["energy_max"] obs[6:8] = feat["tgt_lonlat"] / np.array([180.0, 90.0]) obs[8] = feat["prio"] / 3.0 obs[9] = np.clip(feat["slack"] / 3600.0, 0.0, 1.0) return obs说明：时间用sin/cos编码，避免凌晨0点前后时间值跳变导致的不连续；位置用固定上下界线性压缩，energy和slack压到0~1。slack是tw_end减观测时长再减当前时刻，代表“还有多少富余时间”，对训练晚点惩罚很敏感，值得单独拆出来。归一化上下界必须在生成阶段冻结，训练阶段不允许拿训练集统计量重算，否则验证集特征分布会漂移。
4. 批次排序与数据增强：让训练集不再被场景顺序绑架
生成脚本按场景顺序落盘，如果训练器按文件顺序读，前几十个batch全来自同一批场景，时间相关性极强。批次排序和数据增强是配套动作：排序负责控制分布，增强负责扩大覆盖。
4.1 批次排序的三种组合策略与离线强化学习敏感性
离线强化学习对batch内分布比在线训练敏感得多。以IQL这类离线Q学习方法为例，它不与环境交互，只能从给定batch里估计Q值，batch里高优先级任务占比和全量数据集偏差大，估计出的动作价值就会带偏策略更新。批次排序在这里的作用，是把数据集真实分布重新暴露给优化器。常用组合是三种：
- 按场景规模分桶：以任务数把episode分进S/M/L/XL四档，训练时按桶采样，保证每个batch任务密度一致。
- 按截止时间升序：episode内部任务按tw_end排序，让策略先看到紧迫任务，与人工调度习惯一致。
- 按难度分层：用“平均每任务可访问卫星数”当难度指标，先简单后复杂，形成课程学习。
df["n_tasks"] = df.groupby("episode_id")["task_id"].transform("count") df["bucket"] = pd.cut(df["n_tasks"], [0, 50, 200, 800, 100000], labels=["S", "M", "L", "XL"]) df = df.sort_values(["bucket", "episode_id", "tw_end"])说明：排序键依次是规模桶、场景编号、任务截止时间，这样同一batch内任务密度接近且场景不交叉混入。不要把episode_id的原始编号当排序键，生成顺序里隐含着发射时间、轨道参数的递进变化，按它排序等于把干扰因素也排进batch。
4.2 数据增强的五个操作与合法性约束
选择数据增强方法不要跟图像的风。图像里翻转裁剪是对称性先验，调度数据里能确定的先验只有一条：访问几何可信，任务属性可变。所以我只对任务属性做增强，访问时段保持STK真值不变，既产生新样本又不会引入伪造几何。
| 增强操作 | 具体做法 | 约束条件 |
|---|---|---|
| 时间窗抖动 | tw_start/tw_end整体平移±5% | 不修改访问时段，平移后重叠不足观测时长的样本丢弃 |
| 时长缩放 | 观测时长乘U(0.8,1.2) | 缩放过长导致重叠不足时丢弃 |
| 优先级置换 | 同一episode内prio随机互换 | 只换数值不换任务几何 |
| 卫星掩蔽 | 随机把某颗卫星全部访问置空 | 模拟单星掉线，检验策略冗余 |
| 场景裁剪 | 取任务较少的子集组成新episode | 子集内保持原时间顺序 |
def augment_time_window(df, rng, jitter=0.05): """任务窗口整体平移，访问时段保持不动""" shift = rng.uniform(-1.0, 1.0, len(df)) * df["win_len"] * jitter df = df.copy() df["tw_start"] = df["tw_start"] + shift df["tw_end"] = df["tw_end"] + shift return df def augment_sat_drop(avail, rng, p=0.08): """把指定卫星在全部任务上的可访问窗口置为NaN""" out = avail.copy() for s in range(avail.shape[1]): if rng.random() < p: out[:, s, :] = np.nan return out说明：时间窗抖动的shift按窗口长度比例计算而不是固定秒数，短窗口和长窗口保持各自的紧迫度。卫星掩蔽的p我控制在0.08左右，超过0.2会让“无解场景”占比过高，策略退化成大量拒绝。augment_sat_drop返回三维可用性矩阵，形状是n_task×n_sat×2，第三维存窗口起止，NaN代表不可用。
4.3 增强之后的合法性复检与种子链
增强不是做完就结束，每个操作都可能把可行样本变不可行。复检规则只有一条：重新执行第3章的align逻辑，feasible=False的样本直接丢弃，但丢弃率必须记下来。丢弃率超过5%说明增强参数过猛，训练出来的策略偏保守；丢弃率接近0说明增强没有产生新样本，等于白做。
def recheck(df): before = len(df) ok = df[df["ov_e"] - df["ov_s"] >= df["dur"]] drop_ratio = 1.0 - len(ok) / before return ok, drop_ratio说明：增强、复检、落盘要写进同一个流水线，drop_ratio作为元数据随数据集一起落盘。另外整个流水线里所有rng要用同一条种子链，每个episode的种子由episode_id加全局种子派生。这样中途新增一个增强操作，旧样本仍能按原种子复现，不会让新旧数据集混在一起没法对比。
5. 模型训练与奖励可视化：不让曲线欺骗你
5.1 奖励函数拆成四条可追溯曲线
调度类的奖励往往是加权和，曲线不降时最难判断哪个分量出了问题。不管用DQN还是PPO这类actor-critic结构，做法都一样：把奖励拆成完成、拒绝、晚点、能耗四个分量，分别写进TensorBoard，比只盯episode均值有用得多。
from torch.utils.tensorboard import SummaryWriter writer = SummaryWriter("runs/obs_sched_dqn") writer.add_scalars("reward/decompose", { "complete": ep_complete.mean(), "reject": ep_reject.mean(), "delay": ep_delay.mean(), "energy": ep_energy.mean(), }, step)说明：complete是完成任务的优先级加权和，reject是拒绝惩罚，delay是平均迟到比例，energy是资源消耗。四个量纲不同，add_scalars会为每个键单独建曲线，方便比对。记录频率建议每50个episode一次，每一步都记噪声太大，看不出收敛方向。
5.2 完成率热力图与失败样本回放两个落地技巧
一是画完成率热力图。把目标区域按经纬度分格，统计每个格子的任务完成比例，用matplotlib渲染成图再add_figure进TensorBoard。如果热力图上某些格子和星座覆盖空洞完全重合，说明策略没有学会在无访问时段果断拒绝，而是在做无效尝试。
grid, _, _ = np.histogram2d( tgt_lat, tgt_lon, bins=[36, 72], range=[[-90, 90], [-180, 180]], weights=completed) plt.imshow(grid, origin="lower", aspect="auto") writer.add_figure("accept/heatmap", plt.gcf(), global_step=step)二是失败样本回放。维护一个FIFO队列，存最近若干个低奖励episode，每训练若干步混入这些episode重放。这个技巧本质是带优先级的经验回放，专门治训练中段奖励停滞。最后是部署侧的小技巧：策略网络训好后用torch.onnx.export导出ONNX，调度器用C++侧ONNX Runtime加载，推理延迟压到毫秒级，训练阶段用Python、部署阶段用C++，两套权重不会出现不一致。
本文还有配套的精品资源，点击获取
