添加公式方法

src/preprocessing/process_water_quality_data.py

@@ -4,6 +4,7 @@ if not hasattr(threading.Thread, "isAlive"):
 
 import warnings
 import os
+import re
 import numpy as np
 import pandas as pd
 from scipy import stats
@@ -11,6 +12,54 @@ from scipy import stats
 warnings.filterwarnings("ignore")
 
 
+def auto_detect_coord_columns(df: pd.DataFrame):
+    """
+    双重验证智能识别坐标列：
+    1. 严格正则匹配列名
+    2. 基于数值范围的地理学推断
+    """
+    lon_patterns = [
+        r'^lon', r'^lng', r'^longitude', r'经度', r'^x$', r'^utm_x$', r'^pixel_x$'
+    ]
+    lat_patterns = [
+        r'^lat', r'^latitude', r'纬度', r'^y$', r'^utm_y$', r'^pixel_y$'
+    ]
+
+    x_col, y_col = None, None
+
+    for col in df.columns:
+        col_str = str(col).lower().strip()
+        if x_col is None and any(re.search(p, col_str) for p in lon_patterns):
+            x_col = col
+        if y_col is None and any(re.search(p, col_str) for p in lat_patterns):
+            y_col = col
+
+    if x_col and y_col:
+        return x_col, y_col
+
+    numeric_cols = df.select_dtypes(include=[np.number]).columns.tolist()
+
+    if len(numeric_cols) >= 2:
+        col1, col2 = numeric_cols[0], numeric_cols[1]
+        mean1 = df[col1].head(10).mean()
+        mean2 = df[col2].head(10).mean()
+
+        if abs(mean1) <= 90 and abs(mean2) > 90:
+            y_col, x_col = col1, col2
+        elif abs(mean2) <= 90 and abs(mean1) > 90:
+            x_col, y_col = col1, col2
+        else:
+            if mean1 > mean2:
+                x_col, y_col = col1, col2
+            else:
+                x_col, y_col = col2, col1
+
+        print(f"💡 触发智能数值推断坐标列: X/经度->[{x_col}], Y/纬度->[{y_col}]")
+        return x_col, y_col
+
+    return df.columns[0], df.columns[1]
+
+
 def detect_outliers_iqr(data: pd.DataFrame, column: str) -> pd.Series:
     """使用 IQR 方法检测异常值，返回与 data 同索引的布尔序列"""
     s = pd.to_numeric(data[column], errors="coerce")
@@ -92,11 +141,15 @@ def process_water_quality_data(input_file: str, output_file: str):
     print(f"原始数据形状: {df.shape}")
     print(f"列名: {list(df.columns)}")
 
+    # 0.5) 智能检测坐标列
+    x_col, y_col = auto_detect_coord_columns(df)
+    print(f"坐标列检测结果: X/经度=[{x_col}], Y/纬度=[{y_col}]")
+
     # 1) 经纬度精度筛选（小数位 >= 7）
     print("\n正在筛选经纬度精度(小数位>=7)...")
     initial_count = len(df)
 
-    for col in ["经度", "纬度"]:
+    for col in [y_col, x_col]:
         if col in df.columns:
             dec_len = df[col].apply(_decimal_len)
             keep_mask = dec_len >= 7
@@ -109,26 +162,23 @@ def process_water_quality_data(input_file: str, output_file: str):
 
     # 2) 异常值检测（IQR）- 只删除异常值，不删除整行
     print("\n正在检测异常值(IQR)...")
-    # 数值列
     numeric_columns = df.select_dtypes(include=[np.number]).columns.tolist()
 
-    # 排除不检测的列
-    exclude_columns = ["时间", "测量点", "纬度", "经度"]
+    exclude_columns = {"时间", "测量点", y_col, x_col}
     if "原始" in df.columns:
-        exclude_columns.append("原始")
+        exclude_columns.add("原始")
 
     columns_to_check = [c for c in numeric_columns if c not in exclude_columns]
     print(f"将检测以下列的异常值: {columns_to_check}")
 
     df_clean = df.copy()
     total_outliers_removed = 0
-    
+
     for column in columns_to_check:
         if column in df_clean.columns and df_clean[column].notna().sum() > 0:
             col_mask = detect_outliers_iqr(df_clean, column)
             outlier_count = int(col_mask.sum())
             print(f'列 "{column}" 检测到 {outlier_count} 个异常值，将其设为 NaN')
-            # 只将异常值设为 NaN，不删除整行
             df_clean.loc[col_mask, column] = np.nan
             total_outliers_removed += outlier_count
 
@@ -140,7 +190,7 @@ def process_water_quality_data(input_file: str, output_file: str):
         df_clean = df_clean.drop(columns=["原始"])
         print('已去除 "原始" 列')
 
-    # 4) 字段类型处理：尽量把“时间”转为 datetime
+    # 4) 字段类型处理：尽量把"时间"转为 datetime
     if "时间" in df_clean.columns:
         try:
             df_clean["时间"] = pd.to_datetime(df_clean["时间"], errors="coerce")
@@ -153,22 +203,18 @@ def process_water_quality_data(input_file: str, output_file: str):
         print('错误：未找到 "测量点" 列')
         return
 
-    # 构建聚合字典
     agg_dict = {}
     if "时间" in df_clean.columns and np.issubdtype(df_clean["时间"].dtype, np.datetime64):
-        # 时间取平均（等价于时间戳平均）
         agg_dict["时间"] = "mean"
     elif "时间" in df_clean.columns:
-        # 如果不是时间类型，保留最常见值以避免无意义的字符串平均
         agg_dict["时间"] = lambda s: s.mode().iloc[0] if not s.mode().empty else s.dropna().iloc[0] if s.dropna().size else np.nan
 
-    for col in ["纬度", "经度"]:
+    for col in [y_col, x_col]:
         if col in df_clean.columns:
             agg_dict[col] = "mean"
 
-    # 其余数值列取均值
     for col in df_clean.select_dtypes(include=[np.number]).columns:
-        if col not in ["纬度", "经度"]:
+        if col not in {y_col, x_col}:
             agg_dict[col] = "mean"
 
     grouped = df_clean.groupby("测量点", as_index=False).agg(agg_dict)