Add immutable derivative research inputs and bounded acquisition

docs/DERIVATIVE_RESEARCH.md

@@ -0,0 +1,35 @@
+# 衍生品研究数据包 v1
+
+期货和普通期权共用 `qdk.derivatives/v1`：`manifest.json` 记录来源、用途、限制和 SHA-256；`contracts.json` 是明确的合约规则；`quotes.csv` 是带时区和可用时间的行情。目的目录不可覆盖。只读校验不生成行情或研究结果。
+
+```powershell
+python -m quant_data_kit.derivatives.cli demo --kind future --output demo-futures
+python -m quant_data_kit.derivatives.cli demo --kind option --output demo-options
+python -m quant_data_kit.derivatives.cli validate --bundle demo-options
+```
+
+演示数据包含两个到期月份，期权含三个执行价及看涨/看跌。价格、日历、成交量、保证金均为合成假设，不能据此证明策略有效。
+
+## 合约和行情
+
+合约必需字段：`instrument_id, symbol, product, venue, currency, kind, multiplier, tick, listed_at, last_trade_at, expiry, known_at, timezone, settlement`。时间必须包含时区；`timezone` 为 IANA 时区；币种为三位大写代码。期权另需 `underlying, underlying_kind, option_right, strike, exercise_style`。`initial_margin, maintenance_margin, rules_source` 明确保证金和规则来源。一个包只包含一个合约规则版本；不同历史规则区间分包处理。
+
+CSV 的列顺序见 `Quote` 数据类或演示文件：`instrument_id,at,available_at,session,open,high,low,close,volume,open_interest,settlement,bid,ask,underlying_price`。可选字段为空；价格和金额使用十进制字符串。重复、乱序、未知合约、不一致 OHLC、过期生命周期、NaN、负期权权利金都拒绝。期货数据允许负价用于分析，但当前共享回放只支持正价成交和结算。
+
+`available_at` 表示研究程序能够知道该条观察的最早时间；`known_at` 表示能知道合约规则的时间。补填为历史时间仅是明确假设，不能冒充历史证据。当前包只允许 `synthetic` 或 `retrospective`，不产生 PIT 认证。
+
+## CSV 与 API
+
+```powershell
+python -m quant_data_kit.derivatives.cli import-csv --contracts contracts.json --quotes quotes.csv --output imported --rights-note "my licensed research data" --limits "retrospective rules and availability assumptions"
+python -m quant_data_kit.derivatives.cli fetch --provider dataway --contracts contracts.json --start 2025-01-02 --end 2025-01-10 --output captured --rights-note "authorized internal research"
+python -m quant_data_kit.derivatives.cli fetch --provider databento --dataset GLBX.MDP3 --max-cost-usd 0.25 --contracts contracts.json --start 2025-01-02 --end 2025-01-03 --output captured-db --rights-note "licensed personal research"
+```
+
+- Dataway：管理员设置 `DATAWAY_BASE_URL` 为已有授权网关；不把内部网络地址或密钥写入仓库。只读逐日请求，日期按单日查询定义。空结果不登记为成功。
+- Databento：本地环境变量 `DATABENTO_API_KEY`；先请求费用估算，超过用户填写的上限就停止，下载不自动重试。估算上限不是供应商硬性扣费保证。只允许明确合约 `raw_symbol`，不把连续合约当作可交易合约。
+- 每次限 1–40 个合约、1–32 天，每个响应限 32 MiB；长历史分批采集后按统一规则规范化。
+- 不假设接口有交易所保证金或完整生命周期。Databento OHLCV 不含本接口需要的结算价，因此期货只能先做行情分析；期权缺标的价格时需补齐。API 数据可用时间保守设为次日 UTC 零点并标为回溯假设。
+- 原始响应保留摘要哈希，私有原始市场文件留在本机。规则和价格的真实性由提供者负责；工程校验不授予数据许可。
+
+API 依据：[Databento Historical API](https://databento.com/docs/api-reference-historical?historical=http)。

pyproject.toml

@@ -48,6 +48,7 @@ qdk-providers = "quant_data_kit.cli:main_providers"
 qdk-research-inputs = "quant_data_kit.research_inputs:main"
 qdk-research-dataset = "quant_data_kit.research_dataset:main"
 qdk-financial = "quant_data_kit.financial.cli:main"
+qdk-derivatives = "quant_data_kit.derivatives.cli:main"
 
 [tool.setuptools.packages.find]
 where = ["src"]

src/quant_data_kit/derivatives/__init__.py

@@ -0,0 +1,10 @@
+"""Explicit, immutable inputs for futures and vanilla-option research.
+
+This format validates declared inputs, not exchange authorization or historical
+point-in-time truth. Retrospective imports never acquire a certification label.
+"""
+
+from .bundle import DerivativeBundle, load_bundle, write_bundle
+from .models import Contract, Quote, decimal, utc
+
+__all__ = ["Contract", "DerivativeBundle", "Quote", "decimal", "load_bundle", "utc", "write_bundle"]

src/quant_data_kit/derivatives/bundle.py

@@ -0,0 +1,212 @@
+"""Immutable normalized bundles with source identity and read-only validation."""
+
+from __future__ import annotations
+
+import csv
+import hashlib
+import io
+import json
+import os
+import shutil
+import tempfile
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from pathlib import Path
+
+from .models import Contract, Quote, payload, record, text, utc
+
+SCHEMA = "qdk.derivatives/v1"
+MAX_FILE_BYTES = 128 * 1024 * 1024
+
+
+def json_bytes(value):
+    return (
+        json.dumps(value, sort_keys=True, ensure_ascii=False, indent=2, allow_nan=False) + "\n"
+    ).encode("utf-8")
+
+
+def digest(data):
+    return hashlib.sha256(data).hexdigest()
+
+
+def quote_bytes(quotes):
+    stream = io.StringIO(newline="")
+    writer = csv.DictWriter(stream, list(Quote.__dataclass_fields__), lineterminator="\n")
+    writer.writeheader()
+    writer.writerows(payload(q) for q in quotes)
+    return stream.getvalue().encode("utf-8")
+
+
+def parse_quotes(data):
+    reader = csv.DictReader(io.StringIO(data.decode("utf-8-sig"), newline=""))
+    if reader.fieldnames != list(Quote.__dataclass_fields__):
+        raise ValueError("quote CSV columns must match the documented schema/order")
+    rows = []
+    for raw in reader:
+        if None in raw:
+            raise ValueError("malformed quote CSV row")
+        rows.append(record(Quote, {k: None if v == "" else v for k, v in raw.items()}))
+    return tuple(rows)
+
+
+def validate_records(contracts, quotes):
+    if not contracts or not quotes:
+        raise ValueError("empty derivative dataset is not a successful collection")
+    lookup = {c.instrument_id: c for c in contracts}
+    if len(lookup) != len(contracts):
+        raise ValueError(
+            "duplicate contract IDs; split changed rule versions into separate bundles"
+        )
+    seen = set()
+    previous = {}
+    for q in quotes:
+        if q.instrument_id not in lookup:
+            raise ValueError(f"quote references unknown contract: {q.instrument_id}")
+        c = lookup[q.instrument_id]
+        if not c.listed_at <= q.at <= c.expiry:
+            raise ValueError("quote falls outside the declared contract lifecycle")
+        if c.kind == "option" and min(q.low, q.bid or 0, q.settlement or 0) < 0:
+            raise ValueError("negative option premium")
+        key = (q.instrument_id, q.at)
+        if key in seen or (q.instrument_id in previous and q.at <= previous[q.instrument_id]):
+            raise ValueError("duplicate or out-of-order observations")
+        previous[q.instrument_id] = q.at
+        seen.add(key)
+    if set(previous) != set(lookup):
+        raise ValueError("every declared contract needs at least one observation")
+
+
+@dataclass(frozen=True)
+class DerivativeBundle:
+    root: Path
+    manifest: dict
+    contracts: tuple[Contract, ...]
+    quotes: tuple[Quote, ...]
+    identity: str
+
+    def contract(self, instrument_id):
+        try:
+            return next(c for c in self.contracts if c.instrument_id == instrument_id)
+        except StopIteration as exc:
+            raise ValueError(f"unknown contract: {instrument_id}") from exc
+
+    def asof(self, at, *, product=None, venue=None):
+        at = utc(at)
+        latest = {}
+        contracts = {c.instrument_id: c for c in self.contracts}
+        for q in self.quotes:
+            c = contracts[q.instrument_id]
+            if (
+                q.available_at <= at
+                and c.known_at <= at < c.expiry
+                and (product is None or c.product == product)
+                and (venue is None or c.venue == venue)
+                and (q.instrument_id not in latest or q.at > latest[q.instrument_id].at)
+            ):
+                latest[q.instrument_id] = q
+        return latest
+
+    def verify_unchanged(self):
+        if load_bundle(self.root).identity != self.identity:
+            raise ValueError("input bundle changed during research")
+
+    def summary(self):
+        return {
+            "schema_version": "qdk.derivatives.preflight/v1",
+            "status": "passed",
+            "read_only": True,
+            "evidence_kind": self.manifest["evidence_kind"],
+            "contracts": len(self.contracts),
+            "rows": len(self.quotes),
+            "instrument_ids": [c.instrument_id for c in self.contracts],
+            "currencies": sorted({c.currency for c in self.contracts}),
+            "venues": sorted({c.venue for c in self.contracts}),
+            "input_sha256": self.identity,
+            "limits": self.manifest["limits"],
+        }
+
+
+def load_bundle(root):
+    root = Path(root).resolve()
+    raw_manifest = (root / "manifest.json").read_bytes()
+    if len(raw_manifest) > 1_000_000:
+        raise ValueError("manifest too large")
+    manifest = json.loads(raw_manifest)
+    if not isinstance(manifest, dict):
+        raise ValueError("manifest must be an object")  # noqa: TRY004 - invalid serialized data
+    if manifest.get("schema") != SCHEMA or manifest.get("status") != "complete":
+        raise ValueError("unsupported or incomplete derivative bundle")
+    if manifest.get("evidence_kind") not in {"synthetic", "retrospective"}:
+        raise ValueError(
+            "source must be labeled synthetic or retrospective; no PIT certification implied"
+        )
+    for name in ("provider", "rights_note", "limits"):
+        if not isinstance(manifest.get(name), str) or not manifest[name].strip():
+            raise ValueError(f"{name} must be declared")
+    utc(manifest["captured_at"])
+    files = manifest.get("files", {})
+    if set(files) != {"contracts.json", "quotes.csv"}:
+        raise ValueError("unexpected bundle file set")
+    contents = {}
+    for name, expected in files.items():
+        path = (root / name).resolve()
+        if path.parent != root or not path.is_file() or path.stat().st_size > MAX_FILE_BYTES:
+            raise ValueError("missing, escaped or oversized input file")
+        contents[name] = path.read_bytes()
+        if digest(contents[name]) != expected:
+            raise ValueError(f"input hash mismatch: {name}")
+    raw_contracts = json.loads(contents["contracts.json"])
+    if not isinstance(raw_contracts, list):
+        raise ValueError("contracts must be a list")  # noqa: TRY004 - invalid serialized data
+    contracts = tuple(record(Contract, raw) for raw in raw_contracts)
+    quotes = parse_quotes(contents["quotes.csv"])
+    validate_records(contracts, quotes)
+    identity = digest(raw_manifest + contents["contracts.json"] + contents["quotes.csv"])
+    return DerivativeBundle(root, manifest, contracts, quotes, identity)
+
+
+def write_bundle(
+    output, contracts, quotes, *, provider, evidence_kind, rights_note, limits, source=None
+):
+    contracts, quotes = tuple(contracts), tuple(quotes)
+    validate_records(contracts, quotes)
+    text(provider, "provider")
+    if (
+        evidence_kind not in {"synthetic", "retrospective"}
+        or not rights_note.strip()
+        or not limits.strip()
+    ):
+        raise ValueError("declare source evidence, rights and research limits")
+    output = Path(output).resolve()
+    if output.exists():
+        raise FileExistsError("immutable bundle destination already exists")
+    output.parent.mkdir(parents=True, exist_ok=True)
+    stage = Path(tempfile.mkdtemp(prefix=".derivatives-", dir=output.parent))
+    try:
+        contents = {
+            "contracts.json": json_bytes([payload(c) for c in contracts]),
+            "quotes.csv": quote_bytes(quotes),
+        }
+        for name, data in contents.items():
+            (stage / name).write_bytes(data)
+        manifest = {
+            "schema": SCHEMA,
+            "status": "complete",
+            "provider": provider,
+            "evidence_kind": evidence_kind,
+            "rights_note": rights_note,
+            "limits": limits,
+            "source": source or {},
+            "captured_at": datetime.now(timezone.utc).isoformat(),
+            "symbols": [c.instrument_id for c in contracts],
+            "start": min(q.session for q in quotes),
+            "end": max(q.session for q in quotes),
+            "files": {name: digest(data) for name, data in contents.items()},
+        }
+        (stage / "manifest.json").write_bytes(json_bytes(manifest))
+        load_bundle(stage)
+        os.rename(stage, output)
+    finally:
+        if stage.exists() and stage.resolve().parent == output.parent:
+            shutil.rmtree(stage)
+    return load_bundle(output)

src/quant_data_kit/derivatives/cli.py

@@ -0,0 +1,77 @@
+"""python -m quant_data_kit.derivatives.cli --help"""
+
+import argparse
+import json
+from pathlib import Path
+
+from .bundle import load_bundle, parse_quotes, write_bundle
+from .demo import write_demo
+from .models import Contract, record
+from .providers import collect
+
+
+def main(argv=None):
+    parser = argparse.ArgumentParser(description="Immutable derivative research datasets")
+    sub = parser.add_subparsers(dest="command", required=True)
+    check = sub.add_parser("validate")
+    check.add_argument("--bundle", required=True)
+    demo = sub.add_parser("demo")
+    demo.add_argument("--kind", choices=["future", "option"], required=True)
+    demo.add_argument("--output", required=True)
+    imp = sub.add_parser("import-csv")
+    imp.add_argument("--quotes", required=True, help="normalized quotes.csv")
+    imp.add_argument("--limits", required=True)
+    fetch = sub.add_parser("fetch")
+    fetch.add_argument("--provider", choices=["dataway", "databento"], required=True)
+    fetch.add_argument("--start", required=True)
+    fetch.add_argument("--end", required=True)
+    fetch.add_argument("--dataset")
+    fetch.add_argument("--max-cost-usd")
+    for p in (imp, fetch):
+        p.add_argument("--contracts", required=True, help="explicit contract rules JSON")
+        p.add_argument("--output", required=True)
+        p.add_argument("--rights-note", required=True)
+    args = parser.parse_args(argv)
+    try:
+        if args.command == "validate":
+            result = load_bundle(args.bundle)
+        elif args.command == "demo":
+            result = write_demo(args.output, args.kind)
+        else:
+            contracts = [
+                record(Contract, c)
+                for c in json.loads(Path(args.contracts).read_text(encoding="utf-8-sig"))
+            ]
+            if args.command == "import-csv":
+                result = write_bundle(
+                    args.output,
+                    contracts,
+                    parse_quotes(Path(args.quotes).read_bytes()),
+                    provider="local-csv",
+                    evidence_kind="retrospective",
+                    rights_note=args.rights_note,
+                    limits=args.limits,
+                )
+            else:
+                extra = {}
+                if args.provider == "databento":
+                    if args.max_cost_usd is None or args.dataset is None:
+                        raise ValueError("Databento requires --dataset and --max-cost-usd")
+                    extra.update(dataset=args.dataset, max_cost_usd=args.max_cost_usd)
+                result = collect(
+                    args.output,
+                    contracts,
+                    args.start,
+                    args.end,
+                    provider=args.provider,
+                    rights_note=args.rights_note,
+                    **extra,
+                )
+        print(json.dumps(result.summary(), ensure_ascii=False, indent=2))
+        return 0
+    except (ValueError, OSError, KeyError, TypeError) as exc:
+        parser.exit(2, f"derivative input rejected: {exc}\n")
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())