From bd8b9d0c62114e411a207c93a33cd5d1604b2398 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Wed, 15 Jul 2026 05:03:04 +0800 Subject: [PATCH] feat: add exact sqlite qpk vnext store Co-Authored-By: Codex --- docs/qpk_vnext_n2_sqlite_exact_store.md | 14 ++ .../strategy_lifecycle/qpk_vnext_n2_store.py | 161 ++++++++++++++++++ tests/test_qpk_vnext_n2_store.py | 120 +++++++++++++ 3 files changed, 295 insertions(+) create mode 100644 docs/qpk_vnext_n2_sqlite_exact_store.md create mode 100644 src/quant_platform_kit/strategy_lifecycle/qpk_vnext_n2_store.py create mode 100644 tests/test_qpk_vnext_n2_store.py diff --git a/docs/qpk_vnext_n2_sqlite_exact_store.md b/docs/qpk_vnext_n2_sqlite_exact_store.md new file mode 100644 index 0000000..78ae14c --- /dev/null +++ b/docs/qpk_vnext_n2_sqlite_exact_store.md @@ -0,0 +1,14 @@ +# QPK-N2 SQLite exact foundation + +This slice is a trusted single-process, stdlib-only durable foundation. The +constructor performs no filesystem I/O. `put` is the only operation allowed to +initialize a missing/empty absolute database path; reads never create files. +Existing non-empty databases must already have exactly the v2 marker schema. + +The schema contains only `store_meta(namespace)` and `results(key,payload)`. +Transactions implement create-once, byte-identical idempotency and conflict +without overwrite. `get` accepts a complete N1 key, requires a BLOB-like +payload, decodes and verifies the embedded key, and sanitizes all failures. +Connections close on every success and failure path. SQLite paths are opened +with ordinary `sqlite3.connect(str(path))`, never URI interpolation. Selector, +listing, latest, legacy, migration, network, caller, and live APIs are absent. diff --git a/src/quant_platform_kit/strategy_lifecycle/qpk_vnext_n2_store.py b/src/quant_platform_kit/strategy_lifecycle/qpk_vnext_n2_store.py new file mode 100644 index 0000000..4d76ca3 --- /dev/null +++ b/src/quant_platform_kit/strategy_lifecycle/qpk_vnext_n2_store.py @@ -0,0 +1,161 @@ +"""Exact durable put/get foundation for the trusted qpk-vnext/result/v2 store.""" +from __future__ import annotations + +import json +import os +import sqlite3 +from pathlib import Path +from typing import Any + +from quant_platform_kit.strategy_lifecycle.qpk_vnext_n1 import NAMESPACE, ContractError, ResultContract, decode_wire + + +class StoreError(ValueError): + """Sanitized store failure.""" + + +def _fail() -> None: + raise StoreError("invalid qpk-vnext sqlite store operation") + + +class SqliteResultStore: + """Trusted single-process local store with no selector/latest API.""" + + def __init__(self, db_path: str | os.PathLike[str]) -> None: + if not isinstance(db_path, (str, os.PathLike)) or os.name != "posix": + _fail() + self.db_path = Path(db_path).expanduser() + if not self.db_path.is_absolute() or self.db_path.name in {"", ".", ".."}: + _fail() + + @staticmethod + def _payload(contract: ResultContract) -> bytes: + if not isinstance(contract, ResultContract) or contract.persist_mode != "durable": + _fail() + try: + checked = decode_wire(contract.to_wire()) + return json.dumps(checked.to_wire(), ensure_ascii=False, sort_keys=True, + separators=(",", ":"), allow_nan=False).encode("utf-8") + except (ContractError, TypeError, ValueError, UnicodeError): + _fail() + + @staticmethod + def _schema_names(conn: sqlite3.Connection) -> set[str]: + try: + return {row[0] for row in conn.execute("SELECT name FROM sqlite_master WHERE type='table'")} + except sqlite3.Error: + _fail() + + @staticmethod + def _validate_schema(conn: sqlite3.Connection) -> None: + try: + objects = {(row[0], row[1]) for row in conn.execute( + "SELECT name,type FROM sqlite_master WHERE name NOT LIKE 'sqlite_%'")} + if objects != {("store_meta", "table"), ("results", "table")}: + _fail() + marker = conn.execute("SELECT value FROM store_meta WHERE key=?", ("namespace",)).fetchone() + if marker != (NAMESPACE,): + _fail() + meta_cols = [(row[1], row[2], row[3], row[5]) for row in conn.execute("PRAGMA table_info(store_meta)")] + result_cols = [(row[1], row[2], row[3], row[5]) for row in conn.execute("PRAGMA table_info(results)")] + if meta_cols != [("key", "TEXT", 1, 1), ("value", "TEXT", 1, 0)]: + _fail() + if result_cols != [("key", "TEXT", 1, 1), ("payload", "BLOB", 1, 0)]: + _fail() + except sqlite3.Error: + _fail() + + @staticmethod + def _configure(conn: sqlite3.Connection, *, initialize: bool) -> None: + try: + if initialize: + conn.execute("BEGIN IMMEDIATE") + if SqliteResultStore._schema_names(conn): + SqliteResultStore._validate_schema(conn) + else: + conn.execute("CREATE TABLE store_meta (key TEXT PRIMARY KEY NOT NULL, value TEXT NOT NULL)") + conn.execute("CREATE TABLE results (key TEXT PRIMARY KEY NOT NULL, payload BLOB NOT NULL)") + conn.execute("INSERT INTO store_meta(key,value) VALUES (?,?)", ("namespace", NAMESPACE)) + conn.execute("COMMIT") + else: + SqliteResultStore._validate_schema(conn) + conn.execute("PRAGMA journal_mode=DELETE") + conn.execute("PRAGMA synchronous=FULL") + except (sqlite3.Error, StoreError): + try: + if conn.in_transaction: + conn.execute("ROLLBACK") + except sqlite3.Error: + pass + raise + + def _connect(self, *, write: bool) -> sqlite3.Connection: + conn: sqlite3.Connection | None = None + try: + exists = self.db_path.exists() + if not write and (not exists or self.db_path.stat().st_size == 0): + _fail() + initialize = write and (not exists or self.db_path.stat().st_size == 0) + conn = sqlite3.connect(str(self.db_path), timeout=10, isolation_level=None) + self._configure(conn, initialize=initialize) + return conn + except (OSError, sqlite3.Error, StoreError): + if conn is not None: + conn.close() + _fail() + + @staticmethod + def _key(key: Any) -> str: + if not isinstance(key, str) or not key.startswith(NAMESPACE + "/") or "\\" in key: + _fail() + return key + + @staticmethod + def _blob(value: Any) -> bytes: + if not isinstance(value, (bytes, bytearray, memoryview)): + _fail() + return bytes(value) + + def put(self, contract: ResultContract) -> str: + payload = self._payload(contract) + conn = self._connect(write=True) + try: + conn.execute("BEGIN IMMEDIATE") + row = conn.execute("SELECT payload FROM results WHERE key=?", (contract.key,)).fetchone() + if row is not None: + if self._blob(row[0]) == payload: + conn.execute("COMMIT") + return "idempotent" + conn.execute("ROLLBACK") + _fail() + conn.execute("INSERT INTO results(key,payload) VALUES (?,?)", (contract.key, payload)) + conn.execute("COMMIT") + return "created" + except (sqlite3.Error, StoreError): + try: + if conn.in_transaction: + conn.execute("ROLLBACK") + except sqlite3.Error: + pass + _fail() + finally: + conn.close() + + def get(self, key: str) -> ResultContract: + key = self._key(key) + conn = self._connect(write=False) + try: + row = conn.execute("SELECT payload FROM results WHERE key=?", (key,)).fetchone() + if row is None: + _fail() + data = json.loads(self._blob(row[0]).decode("utf-8")) + item = decode_wire(data) + if item.key != key: + _fail() + if item.persist_mode != "durable": + _fail() + return item + except (sqlite3.Error, StoreError, TypeError, UnicodeError, json.JSONDecodeError): + _fail() + finally: + conn.close() diff --git a/tests/test_qpk_vnext_n2_store.py b/tests/test_qpk_vnext_n2_store.py new file mode 100644 index 0000000..61f5fdb --- /dev/null +++ b/tests/test_qpk_vnext_n2_store.py @@ -0,0 +1,120 @@ +from __future__ import annotations + +import sqlite3 +import json +from concurrent.futures import ThreadPoolExecutor +from dataclasses import replace + +import pytest + +from quant_platform_kit.strategy_lifecycle.qpk_vnext_n1 import ResultContract +from quant_platform_kit.strategy_lifecycle.qpk_vnext_n2_store import SqliteResultStore, StoreError + + +def contract(**changes): + values = dict(domain="us_equity", profile="SOXL", timing="next_open", identity_version=1, + persist_mode="durable", strategy_id="strategy", run_id="run-1", + param_set_id="baseline", param_version=1, source_revision="rev1", + computed_at="2026-07-15T00:00:00.000000Z", params={"x": 1}) + values.update(changes) + return ResultContract(**values) + + +def test_constructor_and_ephemeral_are_side_effect_free(tmp_path): + db = tmp_path / "nested" / "result ?#% 空.sqlite" + store = SqliteResultStore(db) + assert not db.exists() and not db.parent.exists() + with pytest.raises(StoreError): + store.put(replace(contract(), persist_mode="ephemeral")) + assert not db.exists() and not db.parent.exists() + + +def test_exact_write_read_restart_and_idempotent_path_specials(tmp_path): + db = tmp_path / "result ?#% 空.sqlite" + item = contract() + store = SqliteResultStore(db) + assert store.put(item) == "created" + assert SqliteResultStore(db).get(item.key) == item + assert SqliteResultStore(db).put(item) == "idempotent" + with pytest.raises(StoreError): + SqliteResultStore(db).put(contract(computed_at="2026-07-16T00:00:00.000000Z")) + + +def test_concurrent_identical_writers(tmp_path): + store = SqliteResultStore(tmp_path / "results.sqlite") + item = contract() + with ThreadPoolExecutor(max_workers=6) as pool: + outcomes = list(pool.map(lambda _: store.put(item), range(6))) + assert outcomes.count("created") == 1 + assert outcomes.count("idempotent") == 5 + + +def test_missing_wrong_schema_and_payload_corruption_fail_closed(tmp_path): + item = contract() + missing = SqliteResultStore(tmp_path / "missing.sqlite") + with pytest.raises(StoreError): + missing.get(item.key) + assert not (tmp_path / "missing.sqlite").exists() + db = tmp_path / "bad.sqlite" + conn = sqlite3.connect(db) + conn.execute("CREATE TABLE other(value TEXT)"); conn.commit(); conn.close() + before = db.read_bytes() + with pytest.raises(StoreError): + SqliteResultStore(db).get(item.key) + assert db.read_bytes() == before + good = tmp_path / "good.sqlite" + store = SqliteResultStore(good); store.put(item) + conn = sqlite3.connect(good) + conn.execute("UPDATE results SET payload=? WHERE key=?", ("text", item.key)); conn.commit(); conn.close() + with pytest.raises(StoreError): + store.get(item.key) + + +def test_key_mismatch_and_null_payload_and_rollback(tmp_path): + db = tmp_path / "results.sqlite" + item = contract() + conn = sqlite3.connect(db) + conn.execute("CREATE TABLE store_meta (key TEXT PRIMARY KEY, value TEXT NOT NULL)") + conn.execute("CREATE TABLE results (key TEXT PRIMARY KEY, payload BLOB)") + conn.execute("INSERT INTO store_meta(key,value) VALUES ('namespace', 'qpk-vnext/result/v2')") + conn.execute("INSERT INTO results(key,payload) VALUES (?,NULL)", (item.key,)); conn.commit(); conn.close() + with pytest.raises(StoreError): + SqliteResultStore(db).get(item.key) + + +def test_catalog_constraints_and_unexpected_objects_fail_without_mutation(tmp_path): + item = contract() + for suffix, ddl in ( + ("trigger", "CREATE TRIGGER side AFTER INSERT ON results BEGIN SELECT 1; END"), + ("view", "CREATE VIEW extra AS SELECT 1"), + ("index", "CREATE INDEX extra_idx ON results(key)"), + ): + db = tmp_path / f"{suffix}.sqlite" + store = SqliteResultStore(db); store.put(item) + conn = sqlite3.connect(db); conn.execute(ddl); conn.commit(); conn.close() + before = db.read_bytes() + with pytest.raises(StoreError): + store.get(item.key) + assert db.read_bytes() == before + forged = tmp_path / "forged.sqlite" + conn = sqlite3.connect(forged) + conn.execute("CREATE TABLE store_meta (key TEXT PRIMARY KEY NOT NULL, value TEXT NOT NULL)") + conn.execute("CREATE TABLE results (key TEXT PRIMARY KEY, payload BLOB NOT NULL)") + conn.execute("INSERT INTO store_meta(key,value) VALUES ('namespace', 'qpk-vnext/result/v2')") + conn.commit(); conn.close() + with pytest.raises(StoreError): + SqliteResultStore(forged).get(item.key) + + +def test_ephemeral_payload_is_rejected_on_durable_get(tmp_path): + db = tmp_path / "ephemeral.sqlite" + item = contract(persist_mode="ephemeral") + payload = json.dumps(item.to_wire(), sort_keys=True, separators=(",", ":")).encode() + conn = sqlite3.connect(db) + conn.execute("CREATE TABLE store_meta (key TEXT PRIMARY KEY NOT NULL, value TEXT NOT NULL)") + conn.execute("CREATE TABLE results (key TEXT PRIMARY KEY NOT NULL, payload BLOB NOT NULL)") + conn.execute("INSERT INTO store_meta(key,value) VALUES ('namespace', 'qpk-vnext/result/v2')") + conn.execute("INSERT INTO results(key,payload) VALUES (?,?)", (item.key, payload)) + conn.commit(); conn.close() + with pytest.raises(StoreError): + SqliteResultStore(db).get(item.key)