From 3b1684dfb1ecddd0c7df95f1fbac9b21119e9f55 Mon Sep 17 00:00:00 2001 From: PureSaber Date: Sat, 10 Oct 2026 23:58:42 +0800 Subject: [PATCH] Add immutable derivative research bundles and bounded data providers --- docs/DERIVATIVE_RESEARCH.md | 35 ++++ pyproject.toml | 1 + src/quant_data_kit/derivatives/__init__.py | 10 + src/quant_data_kit/derivatives/bundle.py | 212 ++++++++++++++++++++ src/quant_data_kit/derivatives/cli.py | 77 +++++++ src/quant_data_kit/derivatives/demo.py | 106 ++++++++++ src/quant_data_kit/derivatives/models.py | 193 ++++++++++++++++++ src/quant_data_kit/derivatives/providers.py | 197 ++++++++++++++++++ tests/test_derivatives.py | 130 ++++++++++++ 9 files changed, 961 insertions(+) create mode 100644 docs/DERIVATIVE_RESEARCH.md create mode 100644 src/quant_data_kit/derivatives/__init__.py create mode 100644 src/quant_data_kit/derivatives/bundle.py create mode 100644 src/quant_data_kit/derivatives/cli.py create mode 100644 src/quant_data_kit/derivatives/demo.py create mode 100644 src/quant_data_kit/derivatives/models.py create mode 100644 src/quant_data_kit/derivatives/providers.py create mode 100644 tests/test_derivatives.py diff --git a/docs/DERIVATIVE_RESEARCH.md b/docs/DERIVATIVE_RESEARCH.md new file mode 100644 index 0000000..a3e0be3 --- /dev/null +++ b/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)。 diff --git a/pyproject.toml b/pyproject.toml index b8b4c64..34e8593 100644 --- a/pyproject.toml +++ b/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"] diff --git a/src/quant_data_kit/derivatives/__init__.py b/src/quant_data_kit/derivatives/__init__.py new file mode 100644 index 0000000..44e1323 --- /dev/null +++ b/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"] diff --git a/src/quant_data_kit/derivatives/bundle.py b/src/quant_data_kit/derivatives/bundle.py new file mode 100644 index 0000000..0296fe2 --- /dev/null +++ b/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) diff --git a/src/quant_data_kit/derivatives/cli.py b/src/quant_data_kit/derivatives/cli.py new file mode 100644 index 0000000..854a042 --- /dev/null +++ b/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()) diff --git a/src/quant_data_kit/derivatives/demo.py b/src/quant_data_kit/derivatives/demo.py new file mode 100644 index 0000000..79c451a --- /dev/null +++ b/src/quant_data_kit/derivatives/demo.py @@ -0,0 +1,106 @@ +"""Deterministic synthetic markets for engineering acceptance, never market evidence.""" + +from dataclasses import replace +from datetime import timedelta +from decimal import Decimal + +from .bundle import write_bundle +from .models import Contract, Quote, utc + + +def demo_records(kind="future"): + start = utc("2025-01-02T21:00:00Z") + contracts = [] + for month, expiry in [("H", "2025-01-24T21:00:00Z"), ("M", "2025-02-21T21:00:00Z")]: + base = Contract( + instrument_id="DEMO:" + month, + symbol="DEMO" + month, + product="DEMO", + venue="SYNTHETIC", + currency="USD", + kind="future", + multiplier="10", + tick="0.01", + listed_at=start - timedelta(days=100), + last_trade_at=utc(expiry), + expiry=utc(expiry), + known_at=start - timedelta(days=100), + timezone="America/New_York", + settlement="cash", + initial_margin="100", + maintenance_margin="80", + rules_source="Synthetic fixture assumptions; not exchange contract rules", + ) + if kind == "future": + contracts.append(base) + elif kind == "option": + for strike in (95, 100, 105): + for right in ("call", "put"): + code = f"{month}-{right[0].upper()}{strike}" + contracts.append( + replace( + base, + instrument_id="DEMO:" + code, + symbol="DEMO" + code, + kind="option", + underlying="DEMO-SPOT", + option_right=right, + strike=Decimal(strike), + exercise_style="european", + initial_margin=Decimal(300), + maintenance_margin=Decimal(240), + ) + ) + else: + raise ValueError("demo kind must be future or option") + quotes = [] + for i in range(36): + at = start + timedelta(days=i) + if at.weekday() >= 5: + continue + underlying = Decimal(100) + Decimal(i) / 10 + for c in contracts: + if at > c.expiry: + continue + front = c.symbol.startswith("DEMOH") + if kind == "future": + close = underlying + (Decimal("0.20") if front else Decimal("1.20")) + else: + intrinsic = max( + Decimal(0), (underlying - c.strike) * (1 if c.option_right == "call" else -1) + ) + close = intrinsic + Decimal((c.expiry - at).days) / Decimal(25) + previous = max(Decimal(0), close - Decimal("0.10")) + volume = Decimal(1000 - i * 30 if front else 200 + i * 35) + quotes.append( + Quote( + c.instrument_id, + at, + at + timedelta(seconds=1), + str(at.date()), + previous, + close + Decimal("0.20"), + max(Decimal(0), previous - Decimal("0.20")), + close, + max(volume, Decimal(1)), + Decimal(5000), + close, + max(Decimal(0), close - Decimal("0.01")), + close + Decimal("0.01"), + underlying, + ) + ) + return contracts, quotes + + +def write_demo(output, kind): + contracts, quotes = demo_records(kind) + return write_bundle( + output, + contracts, + quotes, + provider="deterministic-demo", + evidence_kind="synthetic", + rights_note="Generated locally; no vendor market data", + limits="Engineering fixture only. Prices, liquidity, calendars and margin are synthetic.", + ) diff --git a/src/quant_data_kit/derivatives/models.py b/src/quant_data_kit/derivatives/models.py new file mode 100644 index 0000000..6368046 --- /dev/null +++ b/src/quant_data_kit/derivatives/models.py @@ -0,0 +1,193 @@ +"""Strict, timezone-aware derivative research records; decimal strings on disk.""" + +from __future__ import annotations + +import re +from dataclasses import asdict, dataclass, fields +from datetime import date, datetime, timezone +from decimal import Decimal, InvalidOperation +from zoneinfo import ZoneInfo, ZoneInfoNotFoundError + + +def decimal(value, name="value") -> Decimal: + if isinstance(value, bool) or not isinstance(value, (str, int, float, Decimal)): + raise ValueError(f"{name}: a finite decimal is required") # noqa: TRY004 - input validation + try: + result = Decimal(str(value)) + except InvalidOperation as exc: + raise ValueError(f"{name}: invalid decimal") from exc + if not result.is_finite() or abs(result) > Decimal("1e18"): + raise ValueError(f"{name}: finite decimal within 1e18 required") + return result + + +def utc(value) -> datetime: + try: + result = ( + value + if isinstance(value, datetime) + else datetime.fromisoformat(str(value).replace("Z", "+00:00")) + ) + except ValueError as exc: + raise ValueError("an ISO timestamp with timezone is required") from exc + if result.tzinfo is None or result.utcoffset() is None: + raise ValueError("timezone-naive timestamp is not allowed") + return result.astimezone(timezone.utc) + + +def text(value, name): + if ( + not isinstance(value, str) + or not value.strip() + or len(value) > 200 + or any(ord(c) < 32 for c in value) + ): + raise ValueError(f"{name}: nonempty text of at most 200 characters required") + return value + + +def record(cls, raw): + if not isinstance(raw, dict) or set(raw) - {f.name for f in fields(cls)}: + raise ValueError(f"unknown fields in {cls.__name__}") + try: + return cls(**raw) + except TypeError as exc: + raise ValueError(f"missing or invalid fields in {cls.__name__}") from exc + + +def payload(item): + return { + k: v.isoformat() + if isinstance(v, (datetime, date)) + else str(v) + if isinstance(v, Decimal) + else v + for k, v in asdict(item).items() + } + + +@dataclass(frozen=True) +class Contract: + instrument_id: str + symbol: str + product: str + venue: str + currency: str + kind: str + multiplier: Decimal + tick: Decimal + listed_at: datetime + last_trade_at: datetime + expiry: datetime + known_at: datetime + timezone: str + settlement: str + underlying: str = "" + underlying_kind: str = "spot" + option_right: str | None = None + strike: Decimal | None = None + exercise_style: str | None = None + initial_margin: Decimal | None = None + maintenance_margin: Decimal | None = None + rules_source: str = "user-declared research assumptions" + + def __post_init__(self): + for name in ("instrument_id", "symbol", "product", "venue", "currency", "rules_source"): + text(getattr(self, name), name) + if not re.fullmatch("[A-Z]{3}", self.currency): + raise ValueError("currency must be a three-letter ISO code") + if self.kind not in {"future", "option"} or self.settlement not in {"cash", "physical"}: + raise ValueError("supported kind: future/option; settlement: cash/physical") + if self.underlying_kind not in {"spot", "future"}: + raise ValueError("underlying_kind must be spot or future") + try: + ZoneInfo(self.timezone) + except (ZoneInfoNotFoundError, TypeError) as exc: + raise ValueError("unknown contract timezone") from exc + for name in ("listed_at", "last_trade_at", "expiry", "known_at"): + object.__setattr__(self, name, utc(getattr(self, name))) + if not self.listed_at <= self.last_trade_at <= self.expiry: + raise ValueError("contract lifecycle dates are inconsistent") + for name in ("multiplier", "tick"): + value = decimal(getattr(self, name), name) + if value <= 0: + raise ValueError(f"{name} must be positive") + object.__setattr__(self, name, value) + for name in ("initial_margin", "maintenance_margin"): + value = getattr(self, name) + if value is not None: + value = decimal(value, name) + if value < 0: + raise ValueError("margin cannot be negative") + object.__setattr__(self, name, value) + if (self.initial_margin is None) != (self.maintenance_margin is None): + raise ValueError("initial and maintenance margin must be provided together") + if self.initial_margin is not None and self.maintenance_margin > self.initial_margin: + raise ValueError("maintenance margin exceeds initial margin") + if self.kind == "option": + text(self.underlying, "underlying") + if self.option_right not in {"call", "put"} or self.exercise_style not in { + "european", + "american", + }: + raise ValueError("option needs call/put and european/american exercise style") + strike = decimal(self.strike, "strike") + if strike <= 0: + raise ValueError("positive strike required for supported vanilla models") + object.__setattr__(self, "strike", strike) + elif any(v is not None for v in (self.option_right, self.strike, self.exercise_style)): + raise ValueError("futures cannot contain option-only fields") + + +@dataclass(frozen=True) +class Quote: + instrument_id: str + at: datetime + available_at: datetime + session: str + open: Decimal + high: Decimal + low: Decimal + close: Decimal + volume: Decimal + open_interest: Decimal | None = None + settlement: Decimal | None = None + bid: Decimal | None = None + ask: Decimal | None = None + underlying_price: Decimal | None = None + + def __post_init__(self): + text(self.instrument_id, "instrument_id") + for name in ("at", "available_at"): + object.__setattr__(self, name, utc(getattr(self, name))) + if self.available_at < self.at: + raise ValueError("quote availability precedes observation") + if date.fromisoformat(self.session).isoformat() != self.session: + raise ValueError("session must be ISO date") + for name in ( + "open", + "high", + "low", + "close", + "volume", + "open_interest", + "settlement", + "bid", + "ask", + "underlying_price", + ): + value = getattr(self, name) + if value is None and name in {"open", "high", "low", "close", "volume"}: + raise ValueError(f"{name}: value is required") + if value is not None: + object.__setattr__(self, name, decimal(value, name)) + if not self.low <= min(self.open, self.close) <= max(self.open, self.close) <= self.high: + raise ValueError("inconsistent OHLC prices") + if self.volume < 0 or (self.open_interest is not None and self.open_interest < 0): + raise ValueError("volume/open interest cannot be negative") + if (self.bid is None) != (self.ask is None) or ( + self.bid is not None and self.bid > self.ask + ): + raise ValueError("both bid and ask required, with bid <= ask") + if self.underlying_price is not None and self.underlying_price <= 0: + raise ValueError("positive underlying required by supported option models") diff --git a/src/quant_data_kit/derivatives/providers.py b/src/quant_data_kit/derivatives/providers.py new file mode 100644 index 0000000..af117f8 --- /dev/null +++ b/src/quant_data_kit/derivatives/providers.py @@ -0,0 +1,197 @@ +"""Bounded read-only acquisition. Contract rules are explicit user inputs. + +Daily vendor bars are retrospective observations. Availability is conservatively +set to the following UTC day; this is an assumption, never historical PIT proof. +No network request is issued on import. No credentials enter bundle provenance. +""" + +from __future__ import annotations + +import base64 +import csv +import io +import json +import os +import re +from datetime import date, datetime, time, timedelta, timezone +from urllib.error import HTTPError, URLError +from urllib.parse import urlencode, urlparse +from urllib.request import Request, urlopen + +from .bundle import digest, write_bundle +from .models import Quote, decimal + +MAX_RESPONSE = 32 * 1024 * 1024 +LIMITS = ( + "Retrospective daily bars; availability assumed at next UTC day, not PIT certified. " + "Contract and margin rules are supplied explicitly; no exchange licensing or rule verification implied." +) + + +def request(url, *, headers=None, data=None): + """No retries on metered requests; redact response bodies and request URLs.""" + try: + with urlopen(Request(url, data=data, headers=headers or {}), timeout=90) as response: + raw = response.read(MAX_RESPONSE + 1) + except HTTPError as exc: + raise ValueError(f"data provider HTTP {exc.code}; check account and request") from None + except (URLError, TimeoutError, OSError): + raise ValueError("data provider unavailable; no bundle was published") from None + if len(raw) > MAX_RESPONSE: + raise ValueError("response exceeds 32 MiB; narrow symbols or dates") + return raw + + +def window(start, end): + start, end = date.fromisoformat(start), date.fromisoformat(end) + if not 0 <= (end - start).days <= 31: + raise ValueError("inclusive acquisition window must be between 1 and 32 days") + return start, end + + +def symbols(contracts): + values = [c.symbol for c in contracts] + if not 1 <= len(values) <= 40 or len(set(values)) != len(values): + raise ValueError("request 1–40 distinct explicit contract symbols") + if any(not re.fullmatch(r"[A-Za-z0-9_. -]{1,64}", s) for s in values): + raise ValueError("invalid contract symbol") + return values + + +def daily_rows(raw, contracts, mapping, *, fixed_day=None, symbol_field="symbol"): + lookup = {c.symbol: c for c in contracts} + reader = csv.DictReader(io.StringIO(raw.decode("utf-8-sig"))) + if not reader.fieldnames or symbol_field not in reader.fieldnames: + raise ValueError("provider response lacks symbol column") + result = [] + for row in reader: + if row.get(symbol_field) not in lookup: + raise ValueError("provider returned an unrequested symbol") + day = fixed_day or date.fromisoformat(row[mapping["session"]][:10]) + at = datetime.combine(day, time(23, 59, 59), timezone.utc) + fields = { + name: row.get(column) or None for name, column in mapping.items() if name != "session" + } + result.append( + Quote( + instrument_id=lookup[row[symbol_field]].instrument_id, + at=at, + available_at=at + timedelta(seconds=1), + session=day.isoformat(), + **fields, + ) + ) + return result + + +def fetch_dataway(contracts, start, end, *, fetch=request, base_url=None): + """Company-authorized Dataway only; returns rows and raw-source hashes.""" + start, end = window(start, end) + base_url = base_url or os.environ.get("DATAWAY_BASE_URL") + parsed = urlparse(base_url or "") + if ( + parsed.scheme not in {"http", "https"} + or not parsed.hostname + or parsed.query + or parsed.fragment + or parsed.username + ): + raise ValueError("configure DATAWAY_BASE_URL with your authorized gateway endpoint") + values = symbols(contracts) + kinds = {c.kind for c in contracts} + if len(kinds) != 1: + raise ValueError("collect futures and options separately") + option = kinds == {"option"} + table = "b_option_marketday" if option else "b_future_globalmarketday" + mapping = {name: f"n_{name}" for name in ("open", "high", "low", "close", "volume")} + mapping.update( + open_interest="n_openInterest", settlement="n_settlement" if option else "n_settlementPrice" + ) + hashes, quotes = [], [] + day = start + while day <= end: + query = { + "tableName": table, + "begDate": day.isoformat(), + "endDate": day.isoformat(), + "fields": ",".join(["c_contract", *mapping.values()]), + "c_contract": ",".join(values), + "n_flag": 1, + } + raw = fetch(base_url + "?" + urlencode(query)) + hashes.append({"session": day.isoformat(), "sha256": digest(raw)}) + quotes.extend(daily_rows(raw, contracts, mapping, fixed_day=day, symbol_field="c_contract")) + day += timedelta(days=1) + return sorted(quotes, key=lambda q: (q.at, q.instrument_id)), { + "table": table, + "raw_responses": hashes, + "date_basis": "one-day provider query", + } + + +def fetch_databento(contracts, start, end, *, dataset, max_cost_usd, fetch=request, key=None): + """Estimate first, then download exact raw-symbol OHLCV-1d request once. + + The budget is a pre-request estimate ceiling, not a provider-enforced spending + cap. Raw symbols must be per-contract: continuous/parent symbology is excluded. + See https://databento.com/docs/api-reference-historical?historical=http. + """ + start, end = window(start, end) + values = symbols(contracts) + budget = decimal(max_cost_usd) + if not 0 <= budget <= 100: + raise ValueError("explicit cost ceiling between 0 and 100 USD required") + if not re.fullmatch(r"[A-Z0-9]+\.[A-Z0-9]+", dataset): + raise ValueError("invalid dataset") + key = key or os.environ.get("DATABENTO_API_KEY") + if not key: + raise ValueError("set DATABENTO_API_KEY in the local environment") + auth = {"Authorization": "Basic " + base64.b64encode((key + ":").encode()).decode()} + query = { + "dataset": dataset, + "symbols": ",".join(values), + "schema": "ohlcv-1d", + "stype_in": "raw_symbol", + "start": start.isoformat(), + "end": (end + timedelta(days=1)).isoformat(), + } + root = "https://hist.databento.com/v0/" + cost = decimal(json.loads(fetch(root + "metadata.get_cost?" + urlencode(query), headers=auth))) + if cost < 0 or cost > budget: + raise ValueError( + f"provider cost estimate {cost} USD exceeds authorized ceiling {budget} USD" + ) + query.update(encoding="csv", pretty_px="true", pretty_ts="true", map_symbols="true") + raw = fetch(root + "timeseries.get_range", headers=auth, data=urlencode(query).encode()) + mapping = {name: name for name in ("open", "high", "low", "close", "volume")} + mapping["session"] = "ts_event" + quotes = daily_rows(raw, contracts, mapping) + if any(not start.isoformat() <= q.session <= end.isoformat() for q in quotes): + raise ValueError("provider returned observations outside requested range") + return sorted(quotes, key=lambda q: (q.at, q.instrument_id)), { + "dataset": dataset, + "schema": "ohlcv-1d", + "raw_sha256": digest(raw), + "estimated_cost_usd": str(cost), + "authorized_ceiling_usd": str(budget), + "date_basis": "UTC daily bar; not exchange settlement session", + } + + +def collect(output, contracts, start, end, *, provider, rights_note, **kwargs): + if provider == "dataway": + quotes, source = fetch_dataway(contracts, start, end, **kwargs) + elif provider == "databento": + quotes, source = fetch_databento(contracts, start, end, **kwargs) + else: + raise ValueError("unsupported derivative provider") + return write_bundle( + output, + contracts, + quotes, + provider=provider, + evidence_kind="retrospective", + rights_note=rights_note, + limits=LIMITS, + source=source, + ) diff --git a/tests/test_derivatives.py b/tests/test_derivatives.py new file mode 100644 index 0000000..a18dbe5 --- /dev/null +++ b/tests/test_derivatives.py @@ -0,0 +1,130 @@ +import json +from dataclasses import replace +from datetime import timedelta +from decimal import Decimal + +import pytest + +from quant_data_kit.derivatives import load_bundle, write_bundle +from quant_data_kit.derivatives.demo import demo_records, write_demo +from quant_data_kit.derivatives.providers import fetch_databento, fetch_dataway + + +@pytest.mark.parametrize("kind", ["future", "option"]) +def test_roundtrip_asof_and_tampering(tmp_path, kind): + bundle = write_demo(tmp_path / kind, kind) + first = bundle.quotes[0] + assert bundle.asof(first.at) == {} + assert first.instrument_id in bundle.asof(first.available_at) + assert bundle.summary()["evidence_kind"] == "synthetic" + assert load_bundle(bundle.root).identity == bundle.identity + with pytest.raises(FileExistsError): + write_demo(bundle.root, kind) + (bundle.root / "quotes.csv").write_text("tampered", encoding="utf-8") + with pytest.raises(ValueError, match="hash mismatch"): + bundle.verify_unchanged() + + +def test_metadata_availability_and_bad_records(tmp_path): + contracts, quotes = demo_records() + c, q = contracts[0], quotes[0] + with pytest.raises(ValueError, match="timezone-naive"): + replace(c, known_at="2025-01-01") + with pytest.raises(ValueError, match="OHLC"): + replace(q, high="1") + with pytest.raises(ValueError, match="required"): + replace(q, close=None) + with pytest.raises(ValueError, match="availability"): + replace(q, available_at=q.at - timedelta(seconds=1)) + with pytest.raises(ValueError, match="finite"): + replace(q, close="NaN") + with pytest.raises(ValueError, match="duplicate"): + write_bundle( + tmp_path / "bad", + [c], + [q, q], + provider="test", + evidence_kind="synthetic", + rights_note="test", + limits="test", + ) + late = replace(c, known_at=q.available_at + timedelta(days=1)) + b = write_bundle( + tmp_path / "late", + [late], + [q], + provider="test", + evidence_kind="synthetic", + rights_note="test", + limits="test", + ) + assert not b.asof(q.available_at) + + +def test_empty_collection_cannot_publish(tmp_path): + contracts, _ = demo_records() + rows, source = fetch_dataway( + contracts, + "2025-01-02", + "2025-01-02", + fetch=lambda url: b"c_contract,n_close\n", + base_url="https://data.example.test/cube", + ) + assert not rows and len(source["raw_responses"]) == 1 + with pytest.raises(ValueError, match="empty"): + write_bundle( + tmp_path / "empty", + contracts, + rows, + provider="dataway", + evidence_kind="retrospective", + rights_note="authorized", + limits="retrospective", + ) + assert not (tmp_path / "empty").exists() + + +def test_paid_provider_cost_gate_and_secret_redaction(): + contracts, _ = demo_records() + calls = [] + + def expensive(url, **kwargs): + calls.append(url) + return b"2.5" + + with pytest.raises(ValueError, match="exceeds"): + fetch_databento( + contracts, + "2025-01-02", + "2025-01-02", + dataset="GLBX.MDP3", + max_cost_usd="1", + key="secret-not-for-provenance", + fetch=expensive, + ) + assert len(calls) == 1 and "metadata.get_cost" in calls[0] + + def allowed(url, **kwargs): + if "metadata.get_cost" in url: + return b"0.1" + assert b"stype_in=raw_symbol" in kwargs["data"] + return b"ts_event,symbol,open,high,low,close,volume\n2025-01-02,DEMOH,100,102,99,101,42\n" + + quotes, source = fetch_databento( + contracts[:1], + "2025-01-02", + "2025-01-02", + dataset="GLBX.MDP3", + max_cost_usd="1", + key="secret", + fetch=allowed, + ) + assert quotes[0].close == Decimal(101) + assert "secret" not in json.dumps(source) + assert quotes[0].available_at.isoformat() == "2025-01-03T00:00:00+00:00" + + +def test_manifest_must_be_object(tmp_path): + (tmp_path / "manifest.json").write_text("[]", encoding="utf-8") + with pytest.raises(ValueError, match="object"): + load_bundle(tmp_path)