严格校验真实行情输入并修复Windows故障注入

src/quant_data_kit/research_dataset.py

@@ -558,8 +558,8 @@ def validate_dataset_frames(
     end: pd.Timestamp,
 ) -> dict[str, Any]:
     """Validate coverage, raw/adjusted identity and evidenced cash adjustments."""
-    validate_price_frame(frames["raw"])
-    validate_price_frame(frames["adjusted"])
+    validate_price_frame(frames["raw"], max_missing_ratio=0.0)
+    validate_price_frame(frames["adjusted"], max_missing_ratio=0.0)
     if not frames["raw"]["adjustment"].eq("none").all():
         raise ValueError("raw prices must be unadjusted")
     if not frames["adjusted"]["adjustment"].eq("qfq").all():

src/quant_data_kit/validate.py

@@ -4,6 +4,7 @@
 
 from collections.abc import Sequence
 
+import numpy as np
 import pandas as pd
 
 from quant_data_kit.exceptions import ValidationError
@@ -26,10 +27,6 @@ def validate_price_frame(
         raise ValidationError("Dataframe is empty")
 
     numeric_cols = [c for c in ("open", "high", "low", "close", "volume") if c in df.columns]
-    missing_ratio = float(df[numeric_cols].isna().mean().mean()) if numeric_cols else 0.0
-    if missing_ratio > max_missing_ratio:
-        raise ValidationError(f"Missing ratio {missing_ratio:.2%} exceeds {max_missing_ratio:.2%}")
-
     dup = (
         df.duplicated(subset=["symbol", "date"]).sum()
         if {"symbol", "date"}.issubset(df.columns)
@@ -38,7 +35,15 @@ def validate_price_frame(
     if dup:
         raise ValidationError(f"Found {dup} duplicate symbol-date rows")
 
-    numeric = df[numeric_cols].apply(pd.to_numeric, errors="coerce")
+    try:
+        numeric = df[numeric_cols].apply(pd.to_numeric, errors="raise")
+    except (TypeError, ValueError) as exc:
+        raise ValidationError("OHLCV values must be numeric") from exc
+    missing_ratio = float(numeric.isna().mean().mean()) if numeric_cols else 0.0
+    if missing_ratio > max_missing_ratio:
+        raise ValidationError(f"Missing ratio {missing_ratio:.2%} exceeds {max_missing_ratio:.2%}")
+    if np.isinf(numeric.to_numpy(dtype=float, na_value=np.nan)).any():
+        raise ValidationError("OHLCV values must be finite")
     if (numeric[[c for c in ("open", "high", "low", "close") if c in numeric]] <= 0).any().any():
         raise ValidationError("OHLC prices must be positive")
     if "volume" in numeric and (numeric["volume"] < 0).any():

tests/process_helpers.py

@@ -0,0 +1,25 @@
+"""Process crash injection without Python, CRT or DLL cleanup."""
+
+from __future__ import annotations
+
+import os
+from typing import NoReturn
+
+
+def hard_exit(code: int) -> NoReturn:
+    """Kill the current worker at the fault point, preserving the exact exit code."""
+    if os.name == "nt":
+        import ctypes
+        from ctypes import wintypes
+
+        # Windows CRT _exit can run DLL_PROCESS_DETACH callbacks. DuckDB can
+        # fault there after its other threads are gone, replacing our exit code.
+        # TerminateProcess models a hard crash without that unrelated teardown.
+        kernel = ctypes.WinDLL("kernel32", use_last_error=True)
+        kernel.GetCurrentProcess.restype = wintypes.HANDLE
+        kernel.TerminateProcess.argtypes = [wintypes.HANDLE, wintypes.UINT]
+        kernel.TerminateProcess.restype = wintypes.BOOL
+        if not kernel.TerminateProcess(kernel.GetCurrentProcess(), code):
+            raise ctypes.WinError(ctypes.get_last_error())
+        raise RuntimeError("TerminateProcess unexpectedly returned")
+    os._exit(code)

tests/test_capture_v2_remediation.py

@@ -44,6 +44,7 @@
 from tests import test_capture_v2_collector_cli_transport as collector_fixtures
 from tests import test_capture_v2_models_storage as storage_fixtures
 from tests import test_capture_v2_synchronizers_epoch as epoch_fixtures
+from tests.process_helpers import hard_exit
 
 
 def _process_publish(
@@ -67,13 +68,13 @@ def _process_crash_after_epoch_snapshot(root_text: str, epoch_id: str) -> None:
 
     def terminate_after_snapshot(*_args: object, **_kwargs: object) -> None:
         (Path(root_text) / "receipt-boundary-reached.marker").write_bytes(b"snapshot-published")
-        os._exit(87)
+        hard_exit(87)
 
     journal._write_committed_receipt = (  # type: ignore[method-assign]
         terminate_after_snapshot
     )
     journal.finalize(created_at=epoch_fixtures.NOW)
-    os._exit(88)
+    hard_exit(88)
 
 
 def test_durable_audit_commits_before_state_and_reloads(tmp_path: Path) -> None:

tests/test_m2_process_integrity.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 import multiprocessing
-import os
 from contextlib import contextmanager
 from copy import deepcopy
 from dataclasses import asdict
@@ -30,6 +29,7 @@
     write_raw_bytes,
 )
 from quant_data_kit.exceptions import ValidationError
+from tests.process_helpers import hard_exit
 from tests.test_adapters_v2 import binance_adapter, cn_adapter, load_messages, okx_adapter
 from tests.test_m2_audit_regressions import trade
 
@@ -57,7 +57,7 @@ def crash(source: Path, destination: Path) -> None:
                 if Path(source).parent.name == ".staging" and Path(destination).name.startswith(
                     "object="
                 ):
-                    os._exit(73)
+                    hard_exit(73)
                 real_replace(source, destination)
 
             lake_module.os.replace = crash
@@ -141,7 +141,7 @@ def _atomic_hard_exit_process(root: str, target_name: str, body: bytes) -> None:
 
     def crash_before_replace(source: Path, destination: Path) -> None:
         if Path(destination) == target:
-            os._exit(71)
+            hard_exit(71)
         real_replace(source, destination)
 
     lake_module.os.replace = crash_before_replace
@@ -158,7 +158,7 @@ def _normalized_hard_exit_process(
 
     def crash_before_publish(source: Path, destination: Path) -> None:
         if Path(destination).parent.name == "snapshots":
-            os._exit(72)
+            hard_exit(72)
         real_replace(source, destination)
 
     lake_module.os.replace = crash_before_publish
@@ -202,7 +202,7 @@ def _curated_hard_exit_process(root: str, normalized_snapshot_id: str) -> None:
 
     def crash_before_publish(source: Path, destination: Path) -> None:
         if Path(destination).parent.name == "snapshots":
-            os._exit(73)
+            hard_exit(73)
         real_replace(source, destination)
 
     lake_module.os.replace = crash_before_publish

tests/test_research_dataset.py

@@ -5,6 +5,7 @@
 import pytest
 
 from quant_data_kit import research_dataset as dataset
+from quant_data_kit.exceptions import ValidationError
 from quant_data_kit.instrument_master import import_instrument_master
 from quant_data_kit.providers.corporate_actions import normalize_etf_actions
 from quant_data_kit.research_coverage import load_history
@@ -109,6 +110,25 @@ def _build(root: Path, source: Path, end="2026-01-21"):
     )
 
 
+@pytest.mark.parametrize("kind", ["raw", "adjusted"])
+@pytest.mark.parametrize("value", [float("nan"), float("inf"), float("-inf"), ""])
+def test_incomplete_volume_cannot_publish_a_research_snapshot(tmp_path, kind, value):
+    source = tmp_path / "source"
+    dates = pd.DatetimeIndex(["2026-01-15", "2026-01-16", "2026-01-19", "2026-01-20", "2026-01-21"])
+    _source(source, dates, [10, 10, 9, 9.1, 9.2], [9, 9, 9, 9.1, 9.2])
+    path = source / f"{kind}.parquet"
+    frame = pd.read_parquet(path)
+    if isinstance(value, str):
+        frame["volume"] = frame["volume"].astype(str)
+    frame.loc[0, "volume"] = value
+    frame.to_parquet(path, index=False)
+    root = tmp_path / "dataset"
+
+    with pytest.raises(ValidationError, match="Missing ratio|must be finite"):
+        _build(root, source)
+    assert not (root / "catalog.json").exists()
+
+
 def test_local_build_is_immutable_asm_compatible_and_records_deferred_cash(tmp_path):
     source = tmp_path / "source"
     dates = pd.DatetimeIndex(["2026-01-15", "2026-01-16", "2026-01-19", "2026-01-20", "2026-01-21"])

tests/test_storage.py

@@ -30,6 +30,23 @@ def test_validate_price_frame_missing_column():
         validate_price_frame(df)
 
 
+@pytest.mark.parametrize("value", [float("inf"), float("-inf"), "not-a-number"])
+def test_validate_price_frame_rejects_invalid_volume(value):
+    frame = pd.DataFrame(
+        {
+            "symbol": ["510300"],
+            "date": ["2026-01-02"],
+            "open": [1.0],
+            "high": [1.2],
+            "low": [0.9],
+            "close": [1.1],
+            "volume": [value],
+        }
+    )
+    with pytest.raises(ValidationError, match="must be finite|must be numeric"):
+        validate_price_frame(frame)
+
+
 def test_cache_covers_range():
     df = pd.DataFrame({"date": pd.date_range("2024-01-01", periods=5)})
     assert cache_covers_range(df, "2024-01-02", "2024-01-04")