"""Validate high-frequency candidates with 5-second raw data. The hourly ``STDDEV`` used by the first pass also reacts to a slow one-sided ramp inside an hour. This module removes a rolling slow trend and counts short reversals, then keeps only candidates that look like repeated up/down movement rather than a load ramp. """ from __future__ import annotations import re from datetime import datetime, timedelta import numpy as np import pandas as pd from scipy.signal import find_peaks from models.common.db import connect DEFAULT_POINT = "YSJ_21" TREND_WINDOW = 121 # 10-minute rolling median at 5s resolution MIN_PROMINENCE = 0.3 # degC; prevent sensor noise from counting MIN_REVERSALS_PER_HOUR = 5.0 MIN_RESIDUAL_STD = 0.5 # degC def _parse(value: str) -> datetime: return datetime.strptime(str(value).strip(), "%Y-%m-%d %H:%M:%S") def _point(row: dict) -> str: point = str(row.get("阶段点位") or DEFAULT_POINT).strip() return point if re.fullmatch(r"YSJ_\d+", point) else DEFAULT_POINT def _metrics(row: dict) -> dict: batch = int(row["批次"]) start = _parse(row["开始小时"]) end = _parse(row["结束小时"]) + timedelta(hours=1) point = _point(row) query_start = start - timedelta(hours=1) query_end = end + timedelta(hours=1) with connect() as connection: with connection.cursor() as cursor: cursor.execute( f"SELECT sample_time, `{point}` AS value " "FROM pks_long_sample " "WHERE import_batch_id = %s AND sample_time >= %s AND sample_time <= %s " "ORDER BY sample_time", (batch, query_start, query_end), ) rows = cursor.fetchall() if not rows: return {"去趋势残差标准差": "", "每小时反复升降次数": "", "是否确认": "否"} times = pd.to_datetime([item["sample_time"] for item in rows]) values = np.asarray([float(item["value"] or 0.0) for item in rows], dtype=float) values[values <= 0] = np.nan valid = ~np.isnan(values) times = times[valid] values = values[valid] if len(values) < 30: return {"去趋势残差标准差": "", "每小时反复升降次数": "", "是否确认": "否"} series = pd.Series(values) trend = series.rolling(TREND_WINDOW, center=True, min_periods=30).median().to_numpy() residual = values - trend window = (times >= start) & (times < end) residual = residual[window] segment_times = times[window] if len(residual) < 30: return {"去趋势残差标准差": "", "每小时反复升降次数": "", "是否确认": "否"} residual_std = float(np.nanstd(residual)) prominence = max(MIN_PROMINENCE, residual_std * 0.7) peaks, _ = find_peaks(residual, prominence=prominence, distance=12) troughs, _ = find_peaks(-residual, prominence=prominence, distance=12) hours = max((segment_times[-1] - segment_times[0]).total_seconds() / 3600.0, 1e-9) reversals = float(len(peaks) + len(troughs)) / hours confirmed = residual_std >= MIN_RESIDUAL_STD and reversals >= MIN_REVERSALS_PER_HOUR row["阶段触发依据"] = ( f"{row.get('阶段触发依据') or ''};去趋势残差{residual_std:.2f}℃、" f"反复升降{reversals:.1f}次/小时" ).strip(";") return { "去趋势残差标准差": round(residual_std, 2), "每小时反复升降次数": round(reversals, 1), "是否确认": "是" if confirmed else "否", } def annotate(hf_rows: list[dict]) -> list[dict]: for row in hf_rows: row.update(_metrics(row)) return hf_rows def filter_hf(hf_rows: list[dict]) -> tuple[list[dict], list[dict]]: """Return ``(confirmed, candidates_with_metrics)``. ``candidates_with_metrics`` keeps every first-pass candidate and its validation metrics; ``confirmed`` only keeps candidates passing the residual/reversal test. """ candidates = annotate(hf_rows) confirmed = [row for row in candidates if row.get("是否确认") == "是"] return confirmed, candidates __all__ = ["annotate", "filter_hf", "MIN_RESIDUAL_STD", "MIN_REVERSALS_PER_HOUR"]