From 1274ae458a7362292752a95f0a612f9d03bb378d Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 27 Jun 2026 16:49:15 +0800 Subject: [PATCH] Add safe dry-run resolver and shared execution marker store. Introduce resolve_dry_run_env with dry-run-safe defaults and move execution marker storage into the kit so equity platforms can share dedup logic. Co-authored-by: Cursor --- pyproject.toml | 2 +- .../common/execution_state.py | 373 ++++++++++++++++++ .../common/runtime_config.py | 13 + tests/test_execution_state.py | 103 +++++ tests/test_runtime_config.py | 5 + 5 files changed, 495 insertions(+), 1 deletion(-) create mode 100644 src/quant_platform_kit/common/execution_state.py create mode 100644 tests/test_execution_state.py diff --git a/pyproject.toml b/pyproject.toml index fccf6983..46f394e4 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "quant-platform-kit" -version = "0.7.38" +version = "0.7.39" description = "Shared broker adapters, domain models, execution ports, and notification utilities for QuantStrategyLab strategies." readme = "README.md" requires-python = ">=3.9" diff --git a/src/quant_platform_kit/common/execution_state.py b/src/quant_platform_kit/common/execution_state.py new file mode 100644 index 00000000..8353b510 --- /dev/null +++ b/src/quant_platform_kit/common/execution_state.py @@ -0,0 +1,373 @@ +"""Execution marker storage for duplicate-run suppression across trading platforms.""" + +from __future__ import annotations + +import json +import re +import tempfile +from collections.abc import Callable, Mapping +from dataclasses import dataclass +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + + +DEFAULT_EXECUTION_STATE_DIR = "/tmp/quant_execution_state" +DEFAULT_EXECUTION_STATE_NAMESPACE = "execution_markers" + + +def _first_non_empty(*values: object) -> str: + for value in values: + text = str(value or "").strip() + if text: + return text + return "" + + +def _env_bool(value: object, *, default: bool) -> bool: + text = str(value or "").strip().lower() + if not text: + return default + if text in {"1", "true", "t", "yes", "y", "on"}: + return True + if text in {"0", "false", "f", "no", "n", "off"}: + return False + return default + + +def _parse_gcs_uri(uri: str) -> tuple[str, str]: + text = str(uri or "").strip() + if not text.startswith("gs://"): + raise ValueError(f"gcs uri must start with gs://, got: {uri!r}") + remainder = text[5:] + bucket, _, prefix = remainder.partition("/") + if not bucket: + raise ValueError(f"gcs uri must include a bucket, got: {uri!r}") + return bucket, prefix.strip("/") + + +def _clean_key_part(value: object, *, fallback: str) -> str: + text = str(value or "").strip().lower() + text = re.sub(r"[^a-z0-9._=-]+", "-", text) + text = re.sub(r"-{2,}", "-", text).strip("-.") + return text or fallback + + +def _clean_relative_key(key: str) -> str: + parts = [ + _clean_key_part(part, fallback="unknown") + for part in str(key or "").replace("\\", "/").split("/") + if str(part or "").strip() + ] + return "/".join(parts) or "unknown" + + +def build_execution_marker_key( + *, + platform: str, + strategy_profile: str, + account_scope: str, + execution_mode: str, + signal_date: object, + effective_date: object, + execution_timing_contract: object = None, +) -> str: + """Build a stable marker key for one strategy signal execution.""" + signal = _first_non_empty(signal_date) + effective = _first_non_empty(effective_date) + if not signal and not effective: + return "" + return "/".join( + ( + "v1", + _clean_key_part(platform, fallback="platform"), + _clean_key_part(account_scope, fallback="account"), + _clean_key_part(strategy_profile, fallback="strategy"), + _clean_key_part(execution_mode, fallback="mode"), + _clean_key_part(signal or "no-signal-date", fallback="signal"), + _clean_key_part(effective or "no-effective-date", fallback="effective"), + _clean_key_part(execution_timing_contract or "no-contract", fallback="contract"), + ) + ) + + +@dataclass(frozen=True) +class ExecutionMarkerStore: + local_dir: str | Path | None = DEFAULT_EXECUTION_STATE_DIR + gcs_prefix_uri: str | None = None + gcp_project_id: str | None = None + namespace: str = DEFAULT_EXECUTION_STATE_NAMESPACE + client_factory: Any = None + prior_report_scan_limit: int = 100 + + def has_marker(self, marker_key: str) -> bool: + if not str(marker_key or "").strip(): + return False + if self.gcs_prefix_uri and self._gcs_blob(marker_key).exists(): + return True + if self.local_dir and self._local_path(marker_key).exists(): + return True + return False + + def record_marker( + self, + marker_key: str, + *, + metadata: Mapping[str, Any] | None = None, + ) -> None: + if not str(marker_key or "").strip(): + return + payload = { + "schema_version": "execution_marker.v1", + "marker_key": str(marker_key), + "recorded_at": datetime.now(timezone.utc).isoformat(), + "metadata": dict(metadata or {}), + } + encoded = json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True) + if self.gcs_prefix_uri: + self._gcs_blob(marker_key).upload_from_string( + encoded, + content_type="application/json", + ) + return + if self.local_dir: + path = self._local_path(marker_key) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(encoded, encoding="utf-8") + + def has_prior_execution_report( + self, + *, + platform: str, + strategy_profile: str, + account_scope: str, + signal_date: object, + effective_date: object, + dry_run_only: bool, + ) -> bool: + if not self.gcs_prefix_uri: + return False + signal = _first_non_empty(signal_date) + effective = _first_non_empty(effective_date) + if not signal and not effective: + return False + month_segment = _month_segment(signal, effective) + bucket_name, prefix = _parse_gcs_uri(str(self.gcs_prefix_uri or "")) + object_prefix = "/".join( + part.strip("/") + for part in ( + prefix, + _runtime_report_segment(platform), + _runtime_report_segment(strategy_profile), + _runtime_report_segment(account_scope), + month_segment, + ) + if part and part.strip("/") + ) + client = self._gcs_client() + scanned = 0 + for blob in client.list_blobs(bucket_name, prefix=object_prefix): + name = str(getattr(blob, "name", "") or "") + if not name.endswith(".json"): + continue + scanned += 1 + if scanned > max(1, int(self.prior_report_scan_limit or 1)): + break + try: + payload = json.loads(blob.download_as_text()) + except Exception: + continue + if _report_matches_execution( + payload, + platform=platform, + strategy_profile=strategy_profile, + account_scope=account_scope, + signal_date=signal, + effective_date=effective, + dry_run_only=dry_run_only, + ): + return True + return False + + def _local_path(self, marker_key: str) -> Path: + root = Path(self.local_dir or tempfile.gettempdir()).expanduser() + return root / self.namespace / f"{_clean_relative_key(marker_key)}.json" + + def _gcs_blob(self, marker_key: str): + bucket_name, prefix = _parse_gcs_uri(str(self.gcs_prefix_uri or "")) + object_name = "/".join( + part.strip("/") + for part in ( + prefix, + self.namespace, + f"{_clean_relative_key(marker_key)}.json", + ) + if part and part.strip("/") + ) + if self.client_factory is None: + try: + from google.cloud import storage # type: ignore + except ImportError as exc: + raise RuntimeError( + "google-cloud-storage is required for GCS execution markers" + ) from exc + client_factory = storage.Client + else: + client_factory = self.client_factory + client = ( + client_factory(project=self.gcp_project_id) + if self.gcp_project_id + else client_factory() + ) + return client.bucket(bucket_name).blob(object_name) + + def _gcs_client(self): + if self.client_factory is None: + try: + from google.cloud import storage # type: ignore + except ImportError as exc: + raise RuntimeError( + "google-cloud-storage is required for GCS execution markers" + ) from exc + client_factory = storage.Client + else: + client_factory = self.client_factory + return ( + client_factory(project=self.gcp_project_id) + if self.gcp_project_id + else client_factory() + ) + + +def build_execution_marker_store_from_env( + *, + platform_env_prefix: str, + env_reader: Callable[[str, str | None], str | None], + gcp_project_id: str | None = None, + client_factory: Any = None, + default_local_dir: str | Path | None = None, +) -> ExecutionMarkerStore: + prefix = str(platform_env_prefix or "").strip().upper() + explicit_gcs_uri = env_reader(f"{prefix}_EXECUTION_STATE_GCS_URI", None) + report_gcs_uri = env_reader("EXECUTION_REPORT_GCS_URI", None) + local_dir = env_reader(f"{prefix}_EXECUTION_STATE_DIR", None) + return ExecutionMarkerStore( + local_dir=local_dir or default_local_dir or DEFAULT_EXECUTION_STATE_DIR, + gcs_prefix_uri=explicit_gcs_uri or report_gcs_uri, + gcp_project_id=gcp_project_id, + client_factory=client_factory, + ) + + +def resolve_execution_dedup_enabled( + *, + platform_env_prefix: str, + env_reader: Callable[[str, str | None], str | None], + dry_run_only: bool, + account_scope: object = None, +) -> bool: + prefix = str(platform_env_prefix or "").strip().upper() + raw_value = env_reader(f"{prefix}_EXECUTION_DEDUP_ENABLED", None) + if raw_value is not None and str(raw_value).strip(): + return _env_bool(raw_value, default=bool(dry_run_only)) + return bool(dry_run_only) or _is_paper_account_scope(account_scope) + + +def _is_paper_account_scope(value: object) -> bool: + return str(value or "").strip().upper() == "PAPER" + + +def _runtime_report_segment(value: object) -> str: + text = str(value or "").strip() + safe = "".join(ch if ch.isalnum() or ch in {"-", "_", "."} else "_" for ch in text) + return safe or "unknown" + + +def _month_segment(*values: object) -> str: + for value in values: + text = _optional_str(value) + if len(text) >= 7 and text[4] == "-" and text[:4].isdigit() and text[5:7].isdigit(): + return text[:7] + return "" + + +def _optional_str(value: object) -> str: + return str(value or "").strip() + + +def _report_matches_execution( + payload: Mapping[str, Any], + *, + platform: str, + strategy_profile: str, + account_scope: str, + signal_date: str, + effective_date: str, + dry_run_only: bool, +) -> bool: + report = dict(payload or {}) + if _optional_str(report.get("platform")).lower() != _optional_str(platform).lower(): + return False + if _optional_str(report.get("strategy_profile")).lower() != _optional_str(strategy_profile).lower(): + return False + if _optional_str(report.get("account_scope")).lower() != _optional_str(account_scope).lower(): + return False + if bool(report.get("dry_run")) != bool(dry_run_only): + return False + summary = dict(report.get("summary") or {}) + if signal_date and _date_key(signal_date) not in _report_signal_date_keys(report, summary): + return False + if effective_date and _date_key(effective_date) not in _report_effective_date_keys(report, summary): + return False + return ( + bool(summary.get("action_done")) + or int(float(summary.get("orders_previewed_count") or 0)) > 0 + or int(float(summary.get("order_events_count") or 0)) > 0 + or _is_successful_no_action_report(report, summary) + ) + + +def _is_successful_no_action_report(report: Mapping[str, Any], summary: Mapping[str, Any]) -> bool: + if _optional_str(report.get("status")).lower() != "ok": + return False + if int(float(summary.get("orders_skipped_count") or 0)) > 0: + return False + return bool("action_done" in summary and not summary.get("action_done")) + + +def _report_signal_date_keys(report: Mapping[str, Any], summary: Mapping[str, Any]) -> set[str]: + signal_snapshot = _report_signal_snapshot(report) + return _date_keys( + summary.get("signal_date"), + signal_snapshot.get("signal_as_of"), + signal_snapshot.get("market_date"), + signal_snapshot.get("price_as_of"), + signal_snapshot.get("snapshot_as_of"), + ) + + +def _report_effective_date_keys(report: Mapping[str, Any], summary: Mapping[str, Any]) -> set[str]: + signal_snapshot = _report_signal_snapshot(report) + return _date_keys( + summary.get("effective_date"), + signal_snapshot.get("effective_date"), + ) + + +def _report_signal_snapshot(report: Mapping[str, Any]) -> dict[str, Any]: + diagnostics = report.get("diagnostics") + if not isinstance(diagnostics, Mapping): + return {} + signal_snapshot = diagnostics.get("signal_snapshot") + return dict(signal_snapshot) if isinstance(signal_snapshot, Mapping) else {} + + +def _date_keys(*values: object) -> set[str]: + return {key for value in values if (key := _date_key(value))} + + +def _date_key(value: object) -> str: + text = _optional_str(value) + if len(text) >= 10 and text[4] == "-" and text[7] == "-": + return text[:10] + return text diff --git a/src/quant_platform_kit/common/runtime_config.py b/src/quant_platform_kit/common/runtime_config.py index 3a2d0d31..907bd271 100644 --- a/src/quant_platform_kit/common/runtime_config.py +++ b/src/quant_platform_kit/common/runtime_config.py @@ -39,6 +39,19 @@ def resolve_bool_value(raw_value: str | None) -> bool: return str(raw_value or "").strip().lower() in {"1", "true", "yes", "y", "on"} +def resolve_dry_run_env( + env: Mapping[str, str | None], + name: str, + *, + default: bool = True, +) -> bool: + """Resolve a dry-run env flag; unset values default to dry-run (safe).""" + raw_value = env.get(name) + if raw_value is None or str(raw_value).strip() == "": + return default + return resolve_bool_value(raw_value) + + def resolve_optional_float_env( env: Mapping[str, str | None], name: str, diff --git a/tests/test_execution_state.py b/tests/test_execution_state.py new file mode 100644 index 00000000..b509a933 --- /dev/null +++ b/tests/test_execution_state.py @@ -0,0 +1,103 @@ +from __future__ import annotations + +import tempfile +import unittest +from pathlib import Path + +from quant_platform_kit.common.execution_state import ( + ExecutionMarkerStore, + build_execution_marker_key, + build_execution_marker_store_from_env, + resolve_execution_dedup_enabled, +) +from quant_platform_kit.common.runtime_config import resolve_dry_run_env + + +class ResolveDryRunEnvTests(unittest.TestCase): + def test_defaults_to_dry_run_when_unset(self) -> None: + self.assertTrue(resolve_dry_run_env({}, "SCHWAB_DRY_RUN_ONLY")) + self.assertTrue(resolve_dry_run_env({"SCHWAB_DRY_RUN_ONLY": ""}, "SCHWAB_DRY_RUN_ONLY")) + self.assertTrue(resolve_dry_run_env({"SCHWAB_DRY_RUN_ONLY": " "}, "SCHWAB_DRY_RUN_ONLY")) + + def test_respects_explicit_false_and_true(self) -> None: + self.assertFalse(resolve_dry_run_env({"SCHWAB_DRY_RUN_ONLY": "false"}, "SCHWAB_DRY_RUN_ONLY")) + self.assertFalse(resolve_dry_run_env({"SCHWAB_DRY_RUN_ONLY": "0"}, "SCHWAB_DRY_RUN_ONLY")) + self.assertTrue(resolve_dry_run_env({"SCHWAB_DRY_RUN_ONLY": "true"}, "SCHWAB_DRY_RUN_ONLY")) + + def test_custom_default(self) -> None: + self.assertFalse(resolve_dry_run_env({}, "SCHWAB_DRY_RUN_ONLY", default=False)) + + +class ExecutionStateTests(unittest.TestCase): + def test_build_execution_marker_key(self) -> None: + key = build_execution_marker_key( + platform="schwab", + strategy_profile="global_etf_rotation", + account_scope="PAPER", + execution_mode="paper", + signal_date="2026-06-01", + effective_date="2026-06-02", + execution_timing_contract="t+1", + ) + self.assertIn("schwab", key) + self.assertIn("global_etf_rotation", key) + self.assertIn("2026-06-01", key) + + def test_local_marker_round_trip(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + store = ExecutionMarkerStore(local_dir=tmpdir, gcs_prefix_uri=None) + key = build_execution_marker_key( + platform="ibkr", + strategy_profile="test", + account_scope="PAPER", + execution_mode="paper", + signal_date="2026-06-01", + effective_date="2026-06-02", + ) + self.assertFalse(store.has_marker(key)) + store.record_marker(key, metadata={"dry_run_only": True}) + self.assertTrue(store.has_marker(key)) + marker_path = Path(tmpdir) / "execution_markers" + self.assertTrue(any(marker_path.iterdir())) + + def test_build_store_from_env(self) -> None: + env = { + "SCHWAB_EXECUTION_STATE_GCS_URI": "gs://bucket/reports", + "SCHWAB_EXECUTION_STATE_DIR": "/tmp/schwab", + } + + def reader(name: str, default: str | None = None) -> str | None: + return env.get(name, default) + + store = build_execution_marker_store_from_env( + platform_env_prefix="SCHWAB", + env_reader=reader, + ) + self.assertEqual(store.gcs_prefix_uri, "gs://bucket/reports") + self.assertEqual(str(store.local_dir), "/tmp/schwab") + + def test_resolve_execution_dedup_enabled(self) -> None: + def reader(name: str, default: str | None = None) -> str | None: + values = {"SCHWAB_EXECUTION_DEDUP_ENABLED": "true"} + return values.get(name, default) + + self.assertTrue( + resolve_execution_dedup_enabled( + platform_env_prefix="SCHWAB", + env_reader=reader, + dry_run_only=False, + account_scope="LIVE", + ) + ) + self.assertTrue( + resolve_execution_dedup_enabled( + platform_env_prefix="SCHWAB", + env_reader=lambda _n, _d=None: None, + dry_run_only=True, + account_scope="LIVE", + ) + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_runtime_config.py b/tests/test_runtime_config.py index ffc5557f..234f4325 100644 --- a/tests/test_runtime_config.py +++ b/tests/test_runtime_config.py @@ -7,6 +7,7 @@ from quant_platform_kit.common.runtime_config import ( first_non_empty, resolve_bool_value, + resolve_dry_run_env, resolve_float_env, resolve_optional_float_env, resolve_quantity_step_env, @@ -31,6 +32,10 @@ def test_common_runtime_config_helpers_normalize_basic_values(self) -> None: self.assertEqual(first_non_empty("", None, " value "), "value") self.assertIs(resolve_bool_value("yes"), True) self.assertIs(resolve_bool_value("0"), False) + self.assertTrue(resolve_dry_run_env({}, "DRY_RUN_ONLY")) + self.assertFalse( + resolve_dry_run_env({"DRY_RUN_ONLY": "false"}, "DRY_RUN_ONLY") + ) env = { "DEFAULT_MIN_NOTIONAL": "25", "EMPTY_MIN_NOTIONAL": "",