From 3c3fd3845d37a9190d35a77172c221805f02be2b Mon Sep 17 00:00:00 2001 From: PureSaber <88314620+PureSaber@users.noreply.github.com> Date: Sun, 11 Oct 2026 11:15:43 +0800 Subject: [PATCH 1/2] Complete institutional research accounts, calibration and acceptance --- README.md | 6 + docs/CORPORATE_EVENTS.md | 32 +++ .../us_research/corporate_events.py | 201 ++++++++++++++++++ 3 files changed, 239 insertions(+) create mode 100644 docs/CORPORATE_EVENTS.md create mode 100644 src/quant_data_kit/us_research/corporate_events.py diff --git a/README.md b/README.md index 13364e1..53432af 100644 --- a/README.md +++ b/README.md @@ -201,3 +201,9 @@ from quant_data_kit.providers.akshare import fetch_hs300_constituents 缺少可核验发布时间的估值记录使用 `available_at=NaT`、`availability_basis=unknown`， 严格 PIT 消费者应拒绝它们。历史指数成分回溯包含查询终点之后、当前快照之前的调整。 升级后应重新抓取旧价格缓存；旧缓存的单位和时间语义不能仅靠升级包自动修复。 + +## 机构研究扩展（2026-10-11） + +追加并购事件修订库与历史可见读取；数据契约见 [CORPORATE_EVENTS](docs/CORPORATE_EVENTS.md)。 + +统一入口与合成输入用法见相邻仓库 [quant-workspace/profiles/institutional-research](../quant-workspace/profiles/institutional-research/README.md)。此增量提供离线合成验收、账户回放和独立 Studio 入口；真实市场数据与权限仍须明确提供。验收命令及边界见上述入口。 diff --git a/docs/CORPORATE_EVENTS.md b/docs/CORPORATE_EVENTS.md new file mode 100644 index 0000000..41dbdc1 --- /dev/null +++ b/docs/CORPORATE_EVENTS.md @@ -0,0 +1,32 @@ +# Corporate events v1 + +`quant_data_kit.us_research.corporate_events` provides `import_events(database, records)` +and `asof_events(database, timestamp)` for a local append-only SQLite store. + +Required fields (unknown fields are rejected): + +| Field | Meaning | +|---|---| +| event_id, revision | Stable deal ID; contiguous positive revisions starting at 1 | +| event_type | `merger` in version 1 | +| target, acquirer | Exact instrument IDs; acquirer may be empty only for cash consideration | +| currency | Three-letter settlement currency | +| announced_at, available_at | Timezone-aware announcement and revision availability; availability cannot precede announcement | +| expected_close_at | Expected completion time, an assumption known with the revision | +| status | announced, pending, completed, terminated | +| cash_per_share, exchange_ratio | Nonnegative decimal strings, with positive total consideration | +| source_ref | Reviewed source URL/document reference, or explicit synthetic reference | +| evidence_kind | synthetic or retrospective; the store does not certify historical PIT | + +Reimporting the same canonical row is idempotent. A conflicting overwrite, noncontiguous +revision, nonincreasing availability, changed identity or reopened terminal event is +rejected transactionally. SHA-256 binds every payload to its database index; all rows +are validated on read. `asof_events` selects the last visible revision per deal, including +visible completed/terminated records. Consumers decide whether these remain eligible. + +The immutable price bundle continues to use `puresaber.us-research/1`. Event terms and +raw prices must use the same share units. Historical accuracy, exhaustive deal coverage, +source licensing, publication timing verification and calibrated probabilities require +additional evidence. No automatic web scraping or financial-news interpretation is included. + +Usage and synthetic-input generation: `quant-workspace/profiles/institutional-research/README.md`. diff --git a/src/quant_data_kit/us_research/corporate_events.py b/src/quant_data_kit/us_research/corporate_events.py new file mode 100644 index 0000000..6a6d51e --- /dev/null +++ b/src/quant_data_kit/us_research/corporate_events.py @@ -0,0 +1,201 @@ +"""Append-only, revision-aware corporate-event research store (SQLite). + +Manual/source-reviewed imports only. A source reference and timestamp are input +claims, not independent historical-PIT certification. No network or order API. +""" + +from __future__ import annotations + +import hashlib +import json +import sqlite3 +from datetime import datetime, timezone +from decimal import Decimal, InvalidOperation +from pathlib import Path + +SCHEMA = "qdk.corporate-events/1" +FIELDS = { + "event_id", + "revision", + "event_type", + "target", + "acquirer", + "currency", + "announced_at", + "available_at", + "status", + "cash_per_share", + "exchange_ratio", + "expected_close_at", + "source_ref", + "evidence_kind", +} + + +def utc(value): + stamp = datetime.fromisoformat(str(value).replace("Z", "+00:00")) + if stamp.tzinfo is None or stamp.utcoffset() is None: + raise ValueError("timezone-aware timestamp required") + return stamp.astimezone(timezone.utc) + + +def number(value, name, *, minimum=None): + if isinstance(value, bool): + raise TypeError(f"{name}: boolean is not numeric") + try: + result = Decimal(str(value)) + except (InvalidOperation, TypeError) as exc: + raise ValueError(f"{name}: decimal required") from exc + if not result.is_finite() or (minimum is not None and result < minimum): + raise ValueError(f"{name}: invalid finite range") + return result + + +def validate(raw): + if not isinstance(raw, dict) or set(raw) != FIELDS: + raise ValueError(f"corporate event requires exactly {sorted(FIELDS)}") + row = dict(raw) + for name in ("event_id", "target", "currency", "source_ref"): + if not isinstance(row[name], str) or not row[name].strip(): + raise ValueError(f"nonempty {name} required") + if len(row["currency"]) != 3 or not row["currency"].isalpha(): + raise ValueError("three-letter currency required") + row["currency"] = row["currency"].upper() + if type(row["revision"]) is not int or row["revision"] < 1: + raise ValueError("positive integer revision required") + if row["event_type"] != "merger": + raise ValueError("v1 supports announced cash/fixed-ratio mergers only") + if row["status"] not in {"announced", "pending", "completed", "terminated"}: + raise ValueError("unsupported merger status") + if row["evidence_kind"] not in {"synthetic", "retrospective"}: + raise ValueError("manual event records cannot certify historical_pit") + for name in ("announced_at", "available_at", "expected_close_at"): + row[name] = utc(row[name]).isoformat() + if utc(row["available_at"]) < utc(row["announced_at"]): + raise ValueError("availability cannot precede announcement") + if utc(row["expected_close_at"]) <= utc(row["announced_at"]): + raise ValueError("expected completion must follow announcement") + for name in ("cash_per_share", "exchange_ratio"): + row[name] = str(number(row[name], name, minimum=0)) + if number(row["cash_per_share"], "cash") + number(row["exchange_ratio"], "ratio") <= 0: + raise ValueError("positive consideration required") + if not isinstance(row["acquirer"], str): + raise TypeError("acquirer must be a string; empty only for cash deals") + if number(row["exchange_ratio"], "ratio") > 0 and not row["acquirer"].strip(): + raise ValueError("stock consideration requires an acquirer instrument") + if row["target"] == row["acquirer"]: + raise ValueError("target and acquirer must differ") + return row + + +def canonical(row): + return json.dumps(row, sort_keys=True, ensure_ascii=False, separators=(",", ":")) + + +def _connect(path, *, readonly=False): + path = Path(path).resolve() + if readonly: + db = sqlite3.connect(path.as_uri() + "?mode=ro", uri=True) + else: + path.parent.mkdir(parents=True, exist_ok=True) + db = sqlite3.connect(path) + db.execute("CREATE TABLE IF NOT EXISTS metadata (schema TEXT NOT NULL)") + if not db.execute("SELECT schema FROM metadata").fetchall(): + db.execute("INSERT INTO metadata VALUES (?)", (SCHEMA,)) + db.execute( + "CREATE TABLE IF NOT EXISTS revisions (event_id TEXT, revision INTEGER, " + "available_at TEXT NOT NULL, payload TEXT NOT NULL, sha256 TEXT NOT NULL, " + "PRIMARY KEY(event_id, revision))" + ) + db.commit() + if db.execute("SELECT schema FROM metadata").fetchall() != [(SCHEMA,)]: + db.close() + raise ValueError("unsupported corporate-event database schema") + return db + + +def _records(db): + result = [] + for event_id, revision, available, payload, digest in db.execute( + "SELECT event_id,revision,available_at,payload,sha256 FROM revisions " + "ORDER BY event_id,revision" + ): + if hashlib.sha256(payload.encode()).hexdigest() != digest: + raise ValueError("corporate-event record checksum mismatch") + row = validate(json.loads(payload)) + if (row["event_id"], row["revision"], row["available_at"]) != ( + event_id, + revision, + available, + ): + raise ValueError("corporate-event index disagrees with payload") + result.append(row) + _validate_history(result) + return result + + +def _validate_history(rows): + previous = {} + for row in sorted(rows, key=lambda r: (r["event_id"], r["revision"])): + old = previous.get(row["event_id"]) + if old: + if row["revision"] != old["revision"] + 1: + raise ValueError("event revisions must be contiguous") + if utc(row["available_at"]) <= utc(old["available_at"]): + raise ValueError("revisions must become available strictly later") + if any( + row[k] != old[k] + for k in ("target", "announced_at", "event_type", "currency", "evidence_kind") + ): + raise ValueError("revision changes event identity") + if old["status"] in {"completed", "terminated"} and row["status"] != old["status"]: + raise ValueError("terminal status cannot reopen; use a new event ID") + elif row["revision"] != 1: + raise ValueError("first event revision must be 1") + previous[row["event_id"]] = row + + +def import_events(path, rows): + rows = [validate(row) for row in rows] + if not rows: + raise ValueError("empty import") + db = _connect(path) + inserted = 0 + try: + with db: + current = {(r["event_id"], r["revision"]): r for r in _records(db)} + for row in rows: + key = (row["event_id"], row["revision"]) + if key in current: + if canonical(current[key]) != canonical(row): + raise ValueError("immutable revision conflict; append a new revision") + continue + current[key] = row + payload = canonical(row) + db.execute( + "INSERT INTO revisions VALUES (?,?,?,?,?)", + ( + *key, + row["available_at"], + payload, + hashlib.sha256(payload.encode()).hexdigest(), + ), + ) + inserted += 1 + _validate_history(current.values()) + finally: + db.close() + return {"schema": SCHEMA, "inserted": inserted, "submitted": len(rows)} + + +def asof_events(path, at): + at = utc(at) + db = _connect(path, readonly=True) + try: + latest = {} + for row in _records(db): + if utc(row["available_at"]) <= at: + latest[row["event_id"]] = row + return [latest[key] for key in sorted(latest)] + finally: + db.close() From f49f1cc306a4a27a2734e21c89ec5b68f7d7384d Mon Sep 17 00:00:00 2001 From: PureSaber <88314620+PureSaber@users.noreply.github.com> Date: Sun, 11 Oct 2026 11:29:40 +0800 Subject: [PATCH 2/2] Cover new modules and validate Windows paths and CI access gates --- .../us_research/corporate_events.py | 11 +- tests/test_corporate_events.py | 125 ++++++++++++++++++ 2 files changed, 135 insertions(+), 1 deletion(-) create mode 100644 tests/test_corporate_events.py diff --git a/src/quant_data_kit/us_research/corporate_events.py b/src/quant_data_kit/us_research/corporate_events.py index 6a6d51e..2e05041 100644 --- a/src/quant_data_kit/us_research/corporate_events.py +++ b/src/quant_data_kit/us_research/corporate_events.py @@ -12,6 +12,7 @@ from datetime import datetime, timezone from decimal import Decimal, InvalidOperation from pathlib import Path +from urllib.parse import quote SCHEMA = "qdk.corporate-events/1" FIELDS = { @@ -95,7 +96,15 @@ def canonical(row): def _connect(path, *, readonly=False): path = Path(path).resolve() if readonly: - db = sqlite3.connect(path.as_uri() + "?mode=ro", uri=True) + # pathlib encodes Windows \\?\ as URI authority %3F, which SQLite rejects. + # SQLite's Windows VFS handles extended filenames after URI decoding. + name = path.as_posix() + if name.startswith("//?/UNC/"): + name = "//" + name[8:] + elif name.startswith("//?/"): + name = name[4:] + uri = "file:" + ("//" if name.startswith("//") else "") + quote(name, safe="/:") + db = sqlite3.connect(uri + "?mode=ro", uri=True) else: path.parent.mkdir(parents=True, exist_ok=True) db = sqlite3.connect(path) diff --git a/tests/test_corporate_events.py b/tests/test_corporate_events.py new file mode 100644 index 0000000..e912d8a --- /dev/null +++ b/tests/test_corporate_events.py @@ -0,0 +1,125 @@ +import sqlite3 + +import pytest + +from quant_data_kit.us_research.corporate_events import ( + asof_events, + import_events, + number, + utc, + validate, +) + + +def event(**changes): + return { + "event_id": "demo", + "revision": 1, + "event_type": "merger", + "target": "T", + "acquirer": "A", + "currency": "usd", + "announced_at": "2025-01-01T00:00:00Z", + "available_at": "2025-01-01T00:00:01Z", + "expected_close_at": "2025-06-01T00:00:00Z", + "status": "announced", + "cash_per_share": "50", + "exchange_ratio": "1", + "source_ref": "synthetic", + "evidence_kind": "synthetic", + **changes, + } + + +def test_append_idempotency_and_asof(tmp_path): + path = tmp_path / "events.sqlite" + first, final = ( + event(), + event(revision=2, available_at="2025-01-02T00:00:00Z", status="completed"), + ) + assert import_events(path, [first, final])["inserted"] == 2 + assert import_events(path, [first])["inserted"] == 0 + assert asof_events(path, "2024-01-01T00:00:00Z") == [] + assert asof_events(path, "2025-01-01T01:00:00Z")[0]["revision"] == 1 + assert asof_events(path, "2025-01-03T00:00:00Z")[0]["status"] == "completed" + with pytest.raises(ValueError, match="conflict"): + import_events(path, [event(cash_per_share="51")]) + + +@pytest.mark.parametrize( + "changes", + [ + {"extra": 1}, + {"event_id": ""}, + {"currency": "12"}, + {"revision": True}, + {"event_type": "split"}, + {"status": "unknown"}, + {"evidence_kind": "historical_pit"}, + {"available_at": "2024-01-01T00:00:00Z"}, + {"expected_close_at": "2024-01-01T00:00:00Z"}, + {"cash_per_share": "0", "exchange_ratio": "0"}, + {"acquirer": ""}, + {"acquirer": "T"}, + {"cash_per_share": "NaN"}, + {"cash_per_share": "-1"}, + {"cash_per_share": "x"}, + ], +) +def test_invalid_event(changes): + with pytest.raises(ValueError): + validate(event(**changes)) + + +def test_invalid_types_and_times(): + with pytest.raises(TypeError): + number(True, "quantity") + with pytest.raises(TypeError): + validate(event(acquirer=1)) + with pytest.raises(ValueError): + utc("2025-01-01") + with pytest.raises(ValueError): + validate([]) + + +@pytest.mark.parametrize( + "changes", + [ + {"revision": 3}, + {"available_at": "2025-01-01T00:00:01Z"}, + {"target": "OTHER"}, + {"status": "pending"}, + ], +) +def test_invalid_history_rolls_back(tmp_path, changes): + path = tmp_path / "events.sqlite" + import_events(path, [event(status="completed")]) + row = event(revision=2, status="completed", available_at="2025-01-02T00:00:00Z") + row.update(changes) + with pytest.raises(ValueError): + import_events(path, [row]) + assert len(asof_events(path, "2025-02-01T00:00:00Z")) == 1 + + +@pytest.mark.parametrize( + "sql", + [ + "UPDATE metadata SET schema='bad'", + "UPDATE revisions SET sha256='bad'", + "UPDATE revisions SET revision=3", + ], +) +def test_tamper_rejected(tmp_path, sql): + path = tmp_path / "events.sqlite" + import_events(path, [event()]) + with sqlite3.connect(path) as db: + db.execute(sql) + with pytest.raises(ValueError): + asof_events(path, "2025-02-01T00:00:00Z") + + +def test_empty_and_noninitial_import(tmp_path): + with pytest.raises(ValueError, match="empty"): + import_events(tmp_path / "e.sqlite", []) + with pytest.raises(ValueError, match="first"): + import_events(tmp_path / "e.sqlite", [event(revision=2)])