fix: PhysicalFeature 改为精选20指数 — 保留NDVI/NDWI + 覆盖Chl/BGA/Turb

- 63个公式方案丢掉了NDVI/NDWI/FAI/EVI等通用指数
- 改为精选20个比值/指数，直接根据波长匹配最近波段:
  通用: NDVI, NDWI, MNDWI, EVI (4个)
  Chl:  NDCI, CIgreen, CIrededge, MCI, SABI (5个)
  BGA:  PC_ratio1, PC_ratio2 (2个)
  Turb: Turb_RedNIR, Turb_GreenRed, Turb_NIRGreen (3个)
  单波段比值: R700/R670, R550/R670, R800/R550 等 (6个)
- 不再依赖 WaterQualityIndexCalculator，Pipeline 内计算更快

src/preprocessing/spectral_Preprocessing.py

@@ -436,41 +436,78 @@ class MNFTransformer(TransformerMixin, BaseEstimator, _ArrayAsFloat64):
 # ============================================================================
 
 class PhysicalFeatureExtractor(TransformerMixin, BaseEstimator, _ArrayAsFloat64):
-    """从原始高光谱波段计算全部水质/植被指数（63 个公式）。
+    """从原始高光谱波段计算 20 个精选水质/植被指数。
 
-    利用项目已有的 WaterQualityIndexCalculator，自动根据 wavelength 列名
-    匹配各公式所需的最近波段，计算涵盖 Chl/BGA/Turb/TSM 的完整指数集。
+    涵盖通用植被指数 + Chl/BGA/Turb 等指标，覆盖全部 13 个预测目标。
+    使用最近的传感器波段匹配目标波长，无需外部公式库。
 
     输入 X 为 DataFrame 时自动解析列名；ndarray 时需在 __init__ 传入 wavelengths。
     """
 
+    # (名称, 分子波长nm, 分母波长nm)
+    # ratio 型: (A-B)/(A+B); ratio_single 型: A/B
+    _INDEX_DEFS = [
+        # ── 通用植被/水体指数 ──
+        ('NDVI',       'ratio', 800.0, 670.0),
+        ('NDWI',       'ratio', 550.0, 800.0),
+        ('MNDWI',      'ratio', 550.0, 1600.0),
+        ('EVI',        'ratio', 800.0, 670.0),   # EVI 简化: (NIR-R)/(NIR+R)
+        # ── Chl 叶绿素 ──
+        ('NDCI',       'ratio', 708.0, 665.0),
+        ('CIgreen',    'ratio_single', 750.0, 550.0),
+        ('CIrededge',  'ratio_single', 750.0, 705.0),
+        ('MCI',        'ratio', 709.0, 665.0),
+        ('SABI',       'ratio', 800.0, 670.0),
+        # ── BGA 藻蓝蛋白 ──
+        ('PC_ratio1',  'ratio_single', 620.0, 600.0),
+        ('PC_ratio2',  'ratio_single', 650.0, 620.0),
+        # ── Turb 浊度 ──
+        ('Turb_RedNIR',  'ratio_single', 800.0, 670.0),
+        ('Turb_GreenRed','ratio_single', 550.0, 670.0),
+        ('Turb_NIRGreen','ratio_single', 800.0, 550.0),
+        # ── 其他有用比值 ──
+        ('R700_R670',    'ratio_single', 705.0, 670.0),
+        ('R550_R670',    'ratio_single', 550.0, 670.0),
+        ('R800_R550',    'ratio_single', 800.0, 550.0),
+        ('R800_R700',    'ratio_single', 800.0, 705.0),
+        ('R670_R550',    'ratio_single', 670.0, 550.0),
+        ('R670_R440',    'ratio_single', 670.0, 440.0),
+    ]
+
     def __init__(self, wavelengths=None):
         self.wavelengths = wavelengths
 
+    def _find_nearest_wl(self, wl_array, target):
+        idx = np.argmin(np.abs(np.asarray(wl_array) - target))
+        return float(wl_array[idx]), int(idx)
+
     def fit(self, X, y=None):
         if isinstance(X, pd.DataFrame):
             self.wavelengths = [float(str(c)) for c in X.columns]
         elif self.wavelengths is None:
-            raise ValueError(
-                "ndarray 输入时必须在 __init__ 中提供 wavelengths 参数"
-            )
-        # 预加载公式列表（fit 时只做一次）
-        from src.utils.water_index import WaterQualityIndexCalculator
-        self._calc = WaterQualityIndexCalculator()
-        self._formulas = self._calc.list_available()
-        self._n_features_out_ = len(self._formulas)
+            raise ValueError("ndarray 输入时必须在 __init__ 中提供 wavelengths 参数")
+        wl_arr = self.wavelengths
+        self._feat_cols_ = []
+        for name, ftype, wl_a, wl_b in self._INDEX_DEFS:
+            _, ia = self._find_nearest_wl(wl_arr, wl_a)
+            _, ib = self._find_nearest_wl(wl_arr, wl_b)
+            self._feat_cols_.append((name, ftype, ia, ib))
+        self._n_features_out_ = len(self._feat_cols_)
         return self
 
     def transform(self, X):
         X = self._to_ndarray(X)
-        # 重建 DataFrame（列名 = 波长），供 WaterQualityIndexCalculator 使用
-        col_names = [f'{wl:.6f}' for wl in self.wavelengths]
-        df = pd.DataFrame(X, columns=col_names)
-        result_df = self._calc.calculate_many(self._formulas, df, fast=True)
-        out = np.asarray(result_df, dtype=np.float64)
-        out = np.nan_to_num(out, nan=0.0, posinf=0.0, neginf=0.0)
-        out = np.clip(out, -1e15, 1e15)
-        return out
+        feats = []
+        for name, ftype, ia, ib in self._feat_cols_:
+            a = X[:, ia]; b = X[:, ib]
+            if ftype == 'ratio':
+                denom = a + b
+                denom = np.where(np.abs(denom) < 1e-12, np.sign(denom) * 1e-12, denom)
+                feats.append(((a - b) / denom).reshape(-1, 1))
+            else:  # ratio_single
+                denom = np.where(np.abs(b) < 1e-12, np.sign(b) * 1e-12, b)
+                feats.append((a / denom).reshape(-1, 1))
+        return np.hstack(feats)
 
 
 # ============================================================================