From 56fbde64926427fa3227c3abd94dccbf0b3ce798 Mon Sep 17 00:00:00 2001 From: claude_dev Date: Sun, 5 Jul 2026 11:53:51 +0800 Subject: [PATCH] =?UTF-8?q?feat(data):=20=E7=A7=BB=E6=A4=8D=20v1=20validat?= =?UTF-8?q?or=20+=20=E9=80=82=E9=85=8D=E6=8E=A5=E5=8F=A3=20+=20=E6=B5=8B?= =?UTF-8?q?=E8=AF=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- sanguo_data/validator.py | 132 +++++++++++++++++++++++++++++++++++ tests/data/conftest.py | 19 +++++ tests/data/test_validator.py | 7 ++ 3 files changed, 158 insertions(+) create mode 100644 sanguo_data/validator.py create mode 100644 tests/data/conftest.py create mode 100644 tests/data/test_validator.py diff --git a/sanguo_data/validator.py b/sanguo_data/validator.py new file mode 100644 index 0000000..6f6ffc5 --- /dev/null +++ b/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] diff --git a/tests/data/conftest.py b/tests/data/conftest.py new file mode 100644 index 0000000..c387191 --- /dev/null +++ b/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], + }) \ No newline at end of file diff --git a/tests/data/test_validator.py b/tests/data/test_validator.py new file mode 100644 index 0000000..a96145e --- /dev/null +++ b/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 \ No newline at end of file