feat(data): 移植 v1 validator + 适配接口 + 测试

sanguo_data/validator.py

@@ -0,0 +1,132 @@
+#!/usr/bin/env python3
+"""数据校验层 - V1 7条fatal规则"""
+import pandas as pd
+from datetime import datetime
+from typing import List, Tuple
+
+
+class ValidationResult:
+    def __init__(self):
+        self.passed = True
+        self.fatal_errors: List[str] = []
+        self.warnings: List[str] = []
+        self.checked_rows = 0
+        self.failed_rows = 0
+
+    def __repr__(self):
+        return (f"ValidationResult(passed={self.passed}, "
+                f"fatal={len(self.fatal_errors)}, warnings={len(self.warnings)}, "
+                f"rows={self.checked_rows}, failed={self.failed_rows})")
+
+    def to_dict(self):
+        return {
+            "passed": self.passed,
+            "fatal_errors": self.fatal_errors,
+            "warnings": self.warnings,
+            "checked_rows": self.checked_rows,
+            "failed_rows": self.failed_rows,
+        }
+
+
+class DataValidator:
+    """数据校验器 - V1 7条fatal规则"""
+
+    def validate(self, df: pd.DataFrame, data_type: str = "daily") -> ValidationResult:
+        result = ValidationResult()
+        if df is None or df.empty:
+            result.fatal_errors.append("数据为空")
+            result.passed = False
+            return result
+        result.checked_rows = len(df)
+
+        if data_type == "daily":
+            self._validate_daily(df, result)
+        elif data_type == "realtime":
+            self._validate_realtime(df, result)
+        return result
+
+    def validate_realtime_dict(self, data: dict) -> ValidationResult:
+        """校验单条实时行情"""
+        result = ValidationResult()
+        result.checked_rows = 1
+        errors = []
+        # R1: 价格>0
+        if not data or data.get("current", 0) <= 0:
+            errors.append("R1: current价格<=0")
+        if data.get("prev_close", 0) <= 0:
+            errors.append("R1: prev_close<=0")
+        # R7: 必须携带source和fetched_at
+        if not data.get("source"):
+            errors.append("R7: 缺少source字段")
+        if not data.get("fetched_at"):
+            errors.append("R7: 缺少fetched_at字段")
+        if errors:
+            result.fatal_errors = errors
+            result.passed = False
+            result.failed_rows = 1
+        return result
+
+    def _validate_daily(self, df: pd.DataFrame, result: ValidationResult):
+        today = datetime.now().strftime("%Y-%m-%d")
+
+        for idx, row in df.iterrows():
+            row_errors = []
+            # D1: 价格>0
+            for col in ["close", "open", "high", "low"]:
+                val = row.get(col, 0)
+                if pd.isna(val) or float(val) <= 0:
+                    row_errors.append(f"D1: {col}<=0 (row {idx})")
+                    break
+
+            # D2: OHLC一致性
+            o, h, l, c = float(row.get("open", 0)), float(row.get("high", 0)), \
+                         float(row.get("low", 0)), float(row.get("close", 0))
+            if o > 0 and c > 0:
+                if h < max(o, c) or l > min(o, c):
+                    row_errors.append(f"D2: OHLC不一致 (row {idx}, o={o} h={h} l={l} c={c})")
+
+            # D3: volume >= 0
+            vol = row.get("volume", 0)
+            if pd.notna(vol) and float(vol) < 0:
+                row_errors.append(f"D3: volume<0 (row {idx})")
+
+            # D7: 非未来日期
+            dt = str(row.get("date", row.get("datetime", "")))[:10]
+            if dt > today:
+                row_errors.append(f"D7: 未来日期 {dt} (row {idx})")
+
+            if row_errors:
+                result.fatal_errors.extend(row_errors)
+                result.failed_rows += 1
+
+        # D6: 日期不重复 (check after all rows)
+        date_col = "date" if "date" in df.columns else "datetime"
+        if date_col in df.columns:
+            dupes = df[df.duplicated(subset=[date_col], keep=False)]
+            if not dupes.empty and len(df) > 1:
+                result.fatal_errors.append(f"D6: {len(dupes)}条重复日期")
+
+        if result.fatal_errors:
+            result.passed = False
+
+
+# 适配层，不改 v1 校验逻辑
+def validate_daily(df):
+    """对外统一接口，委托 v1 校验规则
+
+    Args:
+        df: 输入日线DataFrame
+
+    Returns:
+        过滤后的DataFrame（仅包含通过校验的行）
+    """
+    validator = DataValidator()
+    valid_rows = []
+
+    for idx in range(len(df)):
+        row_df = df.iloc[idx:idx+1]
+        result = validator.validate(row_df, data_type="daily")
+        if result.passed:
+            valid_rows.append(idx)
+
+    return df.iloc[valid_rows]

tests/data/conftest.py

@@ -0,0 +1,19 @@
+import pandas as pd
+import pytest
+
+@pytest.fixture
+def good_daily_df():
+    return pd.DataFrame({
+        "date": ["2026-01-01", "2026-01-02"],
+        "open": [10.0, 11.0], "high": [10.5, 11.5],
+        "low": [9.8, 10.8], "close": [10.2, 11.2],
+        "volume": [10000, 12000],
+    })
+
+@pytest.fixture
+def bad_daily_df():
+    return pd.DataFrame({
+        "date": ["2026-01-01"],
+        "open": [0.0], "high": [0.0], "low": [0.0], "close": [0.0],
+        "volume": [100],
+    })

tests/data/test_validator.py

@@ -0,0 +1,7 @@
+from sanguo_data.validator import validate_daily
+
+def test_validate_daily_keeps_good_rows(good_daily_df):
+    assert len(validate_daily(good_daily_df)) == 2
+
+def test_validate_daily_drops_zero_price(bad_daily_df):
+    assert len(validate_daily(bad_daily_df)) == 0