Add PIT action and halt adapters; fix Windows lake lifecycle

.github/workflows/ci.yml

@@ -11,10 +11,16 @@ permissions:
 
 jobs:
   test:
-    runs-on: ubuntu-latest
+    name: test (${{ matrix.test-label || matrix.python-version }})
+    runs-on: ${{ matrix.os }}
     strategy:
       matrix:
+        os: [ubuntu-latest]
         python-version: ["3.10", "3.11", "3.12"]
+        include:
+          - os: windows-latest
+            python-version: "3.12"
+            test-label: Windows, 3.12
 
     steps:
       - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1

README.md

@@ -1,5 +1,11 @@
 # quant-data-kit
 
+On Windows, deep lake partitions can exceed `MAX_PATH`. If system long paths are
+disabled, supply an absolute extended-length lake root (for example
+`Path("//?/F:/lake")`). The pytest temporary-directory fixture uses this form
+automatically, including in Windows CI; no registry settings are changed.
+Archive file URIs preserve extended-length and UNC roots and literal percent escapes.
+
 研究可信度升级：接口、使用示例、验收及限制见 [11–20 使用说明](docs/SELECTION_LABELS.md)。
 
 Hong Kong daily research uses the explicit `quant_data_kit.hong_kong` module

docs/PUBLIC_CORPORATE_ACTIONS.md

@@ -6,6 +6,14 @@ Ordinary and special CASH dividends on the same ex-date are added only when thei
 
 `providers.tradability.fetch_tradability` captures the current risk-warning board and the requested date's suspension records. `no_reported_restriction` means only that these two responses report no restriction. It does not certify historical listing status, delistings, the next session or price-limit execution. Missing schemas fail rather than turning into an empty restriction list.
 
+`providers.action_bridge.distribution_terms` turns one normalized distribution into separate dividend entitlement, dividend payment and split terms. Rights, mergers and spin-offs are not read from the dividend feed. `explicit_terms` accepts only a complete evidenced record for those kinds.
+
+The bridge preserves `captured_at` as `available_at`; ex/payment dates are economic event dates, not evidence of historical data availability. A later capture must not be backdated for PIT replay.
+
+`providers.halt_calendar.freeze_halt_calendar` stores supplied symbol-session observations and a SHA-256 of the canonical payload. Lookup of a missing session fails. The current Eastmoney ST/halt response cannot be frozen as this calendar.
+
+Every lookup verifies the stored SHA-256 before returning a status. The digest detects changes to the frozen payload; it does not authenticate the source or certify historical completeness.
+
 Primary API field references: [AKShare stock documentation](https://akshare.akfamily.xyz/data/stock/stock.html), sections `stock_dividend_cninfo`, `stock_zh_a_st_em`, and `stock_tfp_em`.
 
 2026-09-19 integration observation: CNInfo returned 109 normalized distribution records across the four example symbols. Eastmoney's current ST endpoint disconnected; no replacement status was fabricated. Tencent's adjusted series and CNInfo cash amounts disagree for the 000333 June 2026 distribution (price adjustment 3.72 versus cash 3.80 per share). Consumers must keep this discrepancy visible; the daily simulation refuses an unexplained adjustment window.

src/quant_data_kit/data_lake.py

@@ -1085,7 +1085,13 @@ def _local_archive_path(archive_uri: str) -> Path:
         path = direct_path
     elif archive_uri.lower().startswith("file:"):
         parsed = urlparse(archive_uri)
-        raw_path = url2pathname(unquote(parsed.path))
+        authority = unquote(parsed.netloc)
+        uri_path = parsed.path
+        if authority and authority.lower() != "localhost":
+            uri_path = "//" + authority + uri_path
+        # url2pathname already percent-decodes. A second decode would turn a
+        # literal %xx filename into a different path and discard UNC/device roots.
+        raw_path = url2pathname(uri_path)
         if os.name == "nt" and re.match(r"^/[A-Za-z]:", raw_path):
             raw_path = raw_path[1:]
         path = Path(raw_path)

src/quant_data_kit/normalized_v3.py

@@ -9,7 +9,7 @@
 import shutil
 import uuid
 from collections.abc import Iterable, Iterator, Mapping, Sequence
-from contextlib import contextmanager
+from contextlib import ExitStack, contextmanager
 from dataclasses import asdict, dataclass, field
 from datetime import date, datetime
 from functools import cache, lru_cache
@@ -526,19 +526,31 @@ def _streaming_stage(root: Path) -> Iterator[Path]:
                     owner.unlink()
             except TimeoutError:
                 continue
-    operation = uuid.uuid4().hex
-    stage = staging_root / f"normalized-batch-stream-{operation}"
-    owner = owners_root / f"{stage.name}.lock"
-    with process_file_lock(owner):
-        stage.mkdir(exist_ok=False)
+        operation = uuid.uuid4().hex
+        stage = staging_root / f"normalized-batch-stream-{operation}"
+        owner = owners_root / f"{stage.name}.lock"
+        ownership = ExitStack()
+        ownership.enter_context(process_file_lock(owner))
         try:
-            yield stage
-        finally:
-            if stage.exists():
-                shutil.rmtree(stage)
-    with process_file_lock(gc_lock):
-        if owner.exists():
+            stage.mkdir(exist_ok=False)
+        except BaseException:
+            ownership.close()
             owner.unlink()
+            raise
+    try:
+        yield stage
+    finally:
+        # Creation and retirement must be atomic with respect to the collector.
+        # Otherwise it can see an owner with no stage, open its lock file, and
+        # race unlink (WinError 32) or split lock identity on POSIX.
+        with process_file_lock(gc_lock):
+            try:
+                if stage.exists():
+                    shutil.rmtree(stage)
+            finally:
+                ownership.close()
+            if owner.exists():
+                owner.unlink()
 
 
 def _sql_text(value: str | Path) -> str:

src/quant_data_kit/providers/action_bridge.py

@@ -0,0 +1,115 @@
+"""Turn evidenced corporate-action rows into execution ActionTerms.
+
+CNInfo distributions already carry cash, bonus shares and transfers. Rights,
+mergers and spin-offs are not inferred from that feed: they enter only through
+an explicit record that satisfies ActionTerms.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from decimal import Decimal
+
+import pandas as pd
+
+from quant_data_kit.financial.actions import KINDS, ActionTerms
+
+_EXPLICIT_KINDS = {"merger", "spin_off", "rights_distribution", "rights_exercise"}
+
+
+def _utc(value, field: str) -> str:
+    stamp = pd.to_datetime(value, errors="coerce")
+    if pd.isna(stamp):
+        raise ValueError(f"corporate action is missing {field}")
+    if stamp.tzinfo is not None:
+        stamp = stamp.tz_convert("UTC")
+    else:
+        stamp = stamp.tz_localize("UTC")
+    return stamp.strftime("%Y-%m-%dT%H:%M:%SZ")
+
+
+def _date(value, field: str) -> str:
+    stamp = pd.to_datetime(value, errors="coerce")
+    if pd.isna(stamp):
+        raise ValueError(f"corporate action is missing {field}")
+    return pd.Timestamp(stamp).strftime("%Y-%m-%d")
+
+
+def distribution_terms(row, *, instrument_id: str, currency: str = "CNY") -> list[ActionTerms]:
+    """Split one normalized CNInfo distribution into dividend and split terms."""
+    cash = Decimal(str(row.cash_per_share))
+    ratio = Decimal(str(row.share_ratio))
+    if not cash.is_finite() or not ratio.is_finite() or cash < 0 or ratio <= 0:
+        raise ValueError("distribution amounts must be non-negative with a positive share ratio")
+    if cash == 0 and ratio == 1:
+        raise ValueError("distribution has neither cash nor a share change")
+    event_id = str(row.event_id)
+    source = str(row.source)
+    record_date = _date(row.record_date, "record_date")
+    ex_at = _utc(row.ex_date, "ex_date")
+    # A historical date in a feed is not evidence that we had the row then.
+    available_at = _utc(row.captured_at, "captured_at")
+    terms: list[ActionTerms] = []
+    if cash > 0:
+        pay_at = _utc(row.pay_date, "pay_date")
+        common = {
+            "instrument_id": instrument_id,
+            "currency": currency,
+            "source": source,
+            "cash_per_unit": str(cash),
+            "entitlement_date": record_date,
+        }
+        terms.append(
+            ActionTerms(
+                event_id=event_id + ":entitlement",
+                kind="dividend_entitlement",
+                effective_at=ex_at,
+                available_at=available_at,
+                evidence_id=event_id,
+                **common,
+            )
+        )
+        terms.append(
+            ActionTerms(
+                event_id=event_id + ":payment",
+                kind="dividend_payment",
+                effective_at=pay_at,
+                available_at=available_at,
+                evidence_id=event_id,
+                **common,
+            )
+        )
+    if ratio != 1:
+        available = _date(row.shares_available_date, "shares_available_date")
+        ex_date = _date(row.ex_date, "ex_date")
+        if available != ex_date:
+            raise ValueError("deferred share delivery is not bridged without a receivables ledger")
+        terms.append(
+            ActionTerms(
+                event_id=event_id + ":split",
+                instrument_id=instrument_id,
+                kind="split",
+                effective_at=ex_at,
+                available_at=available_at,
+                currency=currency,
+                source=source,
+                evidence_id=event_id,
+                ratio=str(ratio),
+            )
+        )
+    return terms
+
+
+def explicit_terms(fields: Mapping) -> ActionTerms:
+    """Accept rights, merger and spin-off terms only when every field is supplied."""
+    if not isinstance(fields, Mapping):
+        raise TypeError("explicit corporate action must be a mapping")
+    kind = fields.get("kind")
+    if kind not in _EXPLICIT_KINDS:
+        raise ValueError(
+            "rights, merger and spin-off require an explicit evidenced record; "
+            f"supported kinds: {sorted(_EXPLICIT_KINDS)}"
+        )
+    if kind not in KINDS:
+        raise ValueError(f"unsupported action kind: {kind}")
+    return ActionTerms(**dict(fields))

src/quant_data_kit/providers/halt_calendar.py

@@ -0,0 +1,73 @@
+"""Content-addressed historical halt calendar.
+
+Sessions that are not in the supplied observations fail closed. A current
+Eastmoney ST/halt snapshot cannot be stored as this calendar.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+
+import pandas as pd
+
+_STATUSES = {"tradable", "suspended", "risk_warning", "delisted"}
+_NOT_TRADABLE = {"suspended", "delisted"}
+_SCHEMA = "quant.halt-calendar/v1"
+
+
+def freeze_halt_calendar(observations: pd.DataFrame, *, source: str) -> dict:
+    if not isinstance(source, str) or not source.strip():
+        raise ValueError("halt calendar source is required")
+    source = source.strip()
+    if source.startswith("akshare.eastmoney.current"):
+        raise ValueError("current ST/halt snapshot cannot certify a historical calendar")
+    required = {"symbol", "session", "status"}
+    if not required.issubset(observations.columns):
+        raise ValueError("halt calendar requires symbol, session and status")
+    rows = []
+    seen: set[tuple[str, str]] = set()
+    for record in observations.to_dict("records"):
+        if pd.isna(record["symbol"]):
+            raise ValueError("halt observation has an empty symbol, date or status")
+        symbol = str(record["symbol"]).strip()
+        session = pd.to_datetime(record["session"], errors="coerce")
+        status = str(record["status"])
+        if not symbol or pd.isna(session) or status not in _STATUSES:
+            raise ValueError("halt observation has an empty symbol, date or status")
+        day = pd.Timestamp(session).strftime("%Y-%m-%d")
+        key = (symbol, day)
+        if key in seen:
+            raise ValueError(f"duplicate halt observation for {symbol} on {day}")
+        seen.add(key)
+        rows.append({"symbol": symbol, "session": day, "status": status})
+    rows.sort(key=lambda item: (item["symbol"], item["session"]))
+    payload = {"schema": _SCHEMA, "source": source, "observations": rows}
+    encoded = json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
+    payload["sha256"] = hashlib.sha256(encoded.encode()).hexdigest()
+    return payload
+
+
+def lookup(calendar: dict, symbol: str, session) -> str:
+    if calendar.get("schema") != _SCHEMA or "sha256" not in calendar:
+        raise ValueError("halt calendar schema is not recognized")
+    payload = {key: value for key, value in calendar.items() if key != "sha256"}
+    encoded = json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
+    if hashlib.sha256(encoded.encode()).hexdigest() != calendar["sha256"]:
+        raise ValueError("halt calendar SHA-256 mismatch")
+    day = pd.Timestamp(session).strftime("%Y-%m-%d")
+    for row in calendar.get("observations", []):
+        if row["symbol"] == symbol and row["session"] == day:
+            if row["status"] not in _STATUSES:
+                raise ValueError("unsupported halt status")
+            return row["status"]
+    raise ValueError(f"session is not covered by the halt calendar: {symbol} {day}")
+
+
+def is_tradable(calendar: dict, symbol: str, session) -> bool:
+    status = lookup(calendar, symbol, session)
+    if status in _NOT_TRADABLE:
+        return False
+    if status in {"tradable", "risk_warning"}:
+        return True
+    raise ValueError(f"unsupported halt status: {status}")

tests/conftest.py

@@ -0,0 +1,15 @@
+"""Exercise deep lake partitions on Windows without a machine-wide registry change."""
+
+import os
+from pathlib import Path
+
+import pytest
+
+
+@pytest.fixture
+def tmp_path(tmp_path: Path) -> Path:
+    if os.name == "nt":
+        path = str(tmp_path.resolve())
+        if not path.startswith("\\\\?\\"):
+            return Path("\\\\?\\UNC\\" + path[2:] if path.startswith("\\\\") else "\\\\?\\" + path)
+    return tmp_path