feat(actpred): 动作历史角度抖动增强——对抗微步样本相关性

微动作插值使相邻样本高度相关，有效多样性远低于样本数（过拟合根因
之一）。--action-jitter >0 时对输入历史加高斯噪声（σ=比例×列物理
范围，如 0.01=舵机±1.8°），制造虚拟邻近状态；且与闭环自预测历史
噪声同源，增强反馈鲁棒性。目标保持干净不抖动。默认 0=关，+2 测试。

actpred/config.py

@@ -33,6 +33,11 @@ class DataConfig:
     # one-hot 软标签高斯宽度（箱）：0=硬标签；>0 真值扩散到邻近箱，
     # 缓解小样本下分类面数据稀疏（1°/箱 × 804 样本多数箱极少被访问）
     onehot_sigma: float = 1.0
+    # 动作历史角度抖动增强（原始量纲比例）：>0 时对输入历史加
+    # 高斯噪声，σ = action_jitter × 列物理范围（舵机 0.01 → ±1.8°）。
+    # 制造虚拟邻近状态，对抗微步插值样本高度相关导致的有效多样性
+    # 不足；且与闭环自预测历史噪声同源，增强反馈鲁棒性。目标不变。
+    action_jitter: float = 0.0
     # R2 方案一：MPU6050 姿态作为独立 IMU token 注入（序列 F+H+2）。
     # 0 = 关闭（默认，序列保持 F+H+1，旧 checkpoint 完全兼容）；
     # 8×N = N 个传感器的 roll,pitch,ax,ay,az,gx,gy,gz 按 pybot 脚本
@@ -119,6 +124,9 @@ def build_argparser(description: str = "") -> argparse.ArgumentParser:
                         help="多头 one-hot 分类输出（bins.py 分箱，舵机 1°/箱）")
     parser.add_argument("--onehot-sigma", type=float, default=None,
                         help="one-hot 软标签高斯宽度（箱，0=硬标签，默认 1.0）")
+    parser.add_argument("--action-jitter", type=float, default=None,
+                        help="动作历史角度抖动（原始量纲比例，0=关；"
+                             "如 0.01 = 舵机 ±1.8°）")
     parser.add_argument("--column-weights", type=str, default=None,
                         help="逐列损失权重（逗号分隔 action_dim 个浮点），"
                              "缺省用默认表（舵机1.0/spk0.2/保留位0.0）；仅影响训练损失")
@@ -214,6 +222,9 @@ def validate_config_values(data_cfg: DataConfig, model_cfg: ModelConfig,
         if train_cfg.predict_delta:
             raise ValueError("onehot 与 predict_delta 互斥"
                              "（残差定义在连续空间）")
+    if not (0.0 <= data_cfg.action_jitter <= 1.0):
+        raise ValueError(
+            f"action_jitter={data_cfg.action_jitter} 应在 [0, 1] 内")
     if not (0.0 <= data_cfg.onehot_sigma <= 10.0):
         raise ValueError(
             f"onehot_sigma={data_cfg.onehot_sigma} 应在 [0, 10] 内")
@@ -333,6 +344,7 @@ def apply_overrides(cfg: Config, args: argparse.Namespace) -> Config:
         "column_weights": ("data", "column_weights"),
         "onehot": ("data", "onehot"),
         "onehot_sigma": ("data", "onehot_sigma"),
+        "action_jitter": ("data", "action_jitter"),
         "imu_dim": ("data", "imu_dim"),
         "backbone": ("model", "cnn_backbone"),
         "cnn_feature_dim": ("model", "cnn_feature_dim"),

actpred/dataset.py

@@ -11,7 +11,8 @@ from torchvision import transforms
 from torchvision.transforms import functional as TF
 
 from imu_norm import imu_scales
-from action_norm import normalize_action, is_range_mode
+from action_norm import (normalize_action, is_range_mode,
+                      action_bounds)
 from bins import encode as bin_encode
 
 
@@ -26,7 +27,7 @@ class ActionDataset(Dataset):
                  normalize_actions: bool = False,
                  action_stats_path: str = None, episode_indices: list = None,
                  frame_len: int = 3, sample_indices: list = None,
-                 augment: bool = False,
+                 augment: bool = False, action_jitter: float = 0.0,
                  predict_delta: bool = False, rollout_len: int = 1,
                  imu_dim: int = 0, normalize_mode: str = "zscore",
                  onehot: bool = False, onehot_sigma: float = 1.0):
@@ -40,6 +41,8 @@ class ActionDataset(Dataset):
         # 多头 one-hot：监督目标为多组 one-hot（bins.py），历史仍走 _norm
         self.onehot = onehot
         self.onehot_sigma = onehot_sigma
+        # 动作历史角度抖动（原始空间加噪，σ ∝ 列物理范围）
+        self.action_jitter = action_jitter
         # R2 方案一：>0 时样本带 "imu" 键（归一化姿态向量，缺失 episode 补零）
         self.imu_dim = imu_dim
         self.image_width = image_width
@@ -133,6 +136,19 @@ class ActionDataset(Dataset):
     def __len__(self):
         return len(self.samples)
 
+    def _jit_history(self, hist_raw):
+        """对原始动作历史加高斯抖动（σ = action_jitter × 列物理范围）
+
+        制造虚拟邻近状态对抗微步样本相关性；与闭环自预测历史噪声同源。
+        目标（真值）不抖动——监督信号保持干净。
+        """
+        if self.action_jitter <= 0.0:
+            return hist_raw
+        lo, hi = action_bounds(self.action_dim)
+        sigma = self.action_jitter * (hi - lo)
+        noise = np.random.normal(0.0, sigma, hist_raw.shape).astype(np.float32)
+        return hist_raw + noise
+
     def _norm(self, a):
         """动作归一化（numpy → torch 前在 numpy 空间做，保持 float32）
 
@@ -212,8 +228,9 @@ class ActionDataset(Dataset):
         if self.rollout_len == 1:
             # 单步：与旧版输出结构完全一致（imu_dim>0 时附加 "imu" 键）
             images = self._load_frames(image_files, t_start)
-            history = torch.tensor(self._norm(
-                actions[t_start - self.history_len: t_start]), dtype=torch.float32)
+            history = torch.tensor(self._norm(self._jit_history(
+                actions[t_start - self.history_len: t_start])),
+                dtype=torch.float32)
             target = torch.tensor(self._step_target(actions, t_start),
                                   dtype=torch.float32)
             out = {"images": images, "action_history": history, "target": target}
@@ -229,7 +246,8 @@ class ActionDataset(Dataset):
                 "images": self._load_frames(image_files, t),
                 # 真实历史（scheduled sampling 的真值来源；首步必用）
                 "action_history": torch.tensor(
-                    self._norm(actions[t - self.history_len: t]), dtype=torch.float32),
+                    self._norm(self._jit_history(
+                        actions[t - self.history_len: t])), dtype=torch.float32),
                 "target": torch.tensor(self._step_target(actions, t), dtype=torch.float32),
                 # 上一步真实动作（Δ 残差的基准 a_{t-1}，归一化空间）
                 "prev_action": torch.tensor(self._norm(actions[t - 1]), dtype=torch.float32),

actpred/test_actpred.py

@@ -2525,6 +2525,56 @@ class TestOneHot(unittest.TestCase):
         cfg2 = apply_overrides(Config(), parser.parse_args([]))
         self.assertEqual(cfg2.train.val_split_mode, "sample")  # 新默认
 
+    def test_action_jitter(self):
+        """角度抖动增强：历史加有界高斯噪声，目标保持干净"""
+        with tempfile.TemporaryDirectory() as tmpdir:
+            base = Path(tmpdir)
+            ep_dir = base / "episode_0001"
+            images_dir = ep_dir / "images"
+            images_dir.mkdir(parents=True)
+            for i in range(10):
+                Image.fromarray(np.random.randint(
+                    0, 255, (16, 16, 3), dtype=np.uint8)).save(
+                    images_dir / f"{i:04d}.jpg")
+            actions = np.zeros((10, 4), dtype=np.float32)
+            actions[:, 0] = 90.0
+            np.save(ep_dir / "actions.npy", actions)
+            # jitter=0：历史 = 原始归一化值
+            ds0 = ActionDataset(tmpdir, history_len=3, action_dim=4,
+                                image_width=16, image_height=16, frame_len=3,
+                                normalize_actions=True,
+                                normalize_mode="fixed_range",
+                                action_jitter=0.0)
+            np.testing.assert_allclose(
+                ds0[3]["action_history"].numpy()[:, 0], 0.5, atol=1e-6)
+            # jitter>0：历史围绕 0.5 抖动（0.1×180=18°→归一化 σ=0.1），
+            # 目标仍是干净的 0.5
+            np.random.seed(42)
+            dsj = ActionDataset(tmpdir, history_len=3, action_dim=4,
+                                image_width=16, image_height=16, frame_len=3,
+                                normalize_actions=True,
+                                normalize_mode="fixed_range",
+                                action_jitter=0.1)
+            hist = dsj[3]["action_history"].numpy()[:, 0]
+            tgt = dsj[3]["target"].numpy()[0]
+            self.assertGreater(float(np.std(hist)), 0.01)   # 确实抖了
+            self.assertLess(float(np.abs(hist - 0.5).max()), 0.45)  # 有界
+            self.assertAlmostEqual(float(tgt), 0.5, places=6)  # 目标干净
+            # 同一索引两次采样抖动不同（每次 __getitem__ 重新加噪）
+            hist2 = dsj[3]["action_history"].numpy()[:, 0]
+            self.assertFalse(np.allclose(hist, hist2))
+
+    def test_action_jitter_cli(self):
+        """--action-jitter 参数映射与校验"""
+        from config import build_argparser, apply_overrides
+        cfg = apply_overrides(Config(),
+                              build_argparser("t").parse_args(
+                                  ["--action-jitter", "0.02"]))
+        self.assertEqual(cfg.data.action_jitter, 0.02)
+        cfg.data.action_jitter = 1.5
+        with self.assertRaises(ValueError):
+            validate_config_values(cfg.data, cfg.model, cfg.train)
+
     def test_onehot_config_validation(self):
         cfg = Config()
         cfg.data.onehot = True

actpred/train.py

@@ -239,7 +239,8 @@ def main():
         augment=cfg.train.augment,
         predict_delta=cfg.train.predict_delta, rollout_len=cfg.train.rollout_len,
         imu_dim=cfg.data.imu_dim, normalize_mode=cfg.data.normalize_mode,
-        onehot=cfg.data.onehot, onehot_sigma=cfg.data.onehot_sigma)
+        onehot=cfg.data.onehot, onehot_sigma=cfg.data.onehot_sigma,
+        action_jitter=cfg.data.action_jitter)
     auto_stats = False
     if (cfg.data.normalize_actions and cfg.data.normalize_mode == "zscore"
             and not cfg.data.action_stats_path):
@@ -258,7 +259,8 @@ def main():
         sample_indices=val_idx if cfg.train.val_split_mode == "sample" else None,
         predict_delta=cfg.train.predict_delta, rollout_len=cfg.train.rollout_len,
         imu_dim=cfg.data.imu_dim, normalize_mode=cfg.data.normalize_mode,
-        onehot=cfg.data.onehot, onehot_sigma=cfg.data.onehot_sigma)
+        onehot=cfg.data.onehot, onehot_sigma=cfg.data.onehot_sigma,
+        action_jitter=cfg.data.action_jitter)
     if auto_stats:
         # checkpoint 配置只存裸文件名：推理端按 checkpoint 目录解析相对路径
         cfg.data.action_stats_path = "action_stats.npy"