diff --git a/src/quant_strategy_plugins/tqqq_market_regime_control_present.py b/src/quant_strategy_plugins/tqqq_market_regime_control_present.py new file mode 100644 index 0000000..18dc786 --- /dev/null +++ b/src/quant_strategy_plugins/tqqq_market_regime_control_present.py @@ -0,0 +1,537 @@ +from __future__ import annotations + +import argparse +import base64 +from dataclasses import dataclass +from datetime import date +import hashlib +import json +import math +import os +from pathlib import Path +import shutil +import subprocess +import sys +import tempfile +from typing import Any, Mapping + +from .strategy_plugin_runner import _flatten_strategy_plugin_entry, load_plugin_config, run_market_regime_control_plugin + + +SCHEMA = "qsl.tqqq_market_regime_control_present.v1" +PRODUCER_ENTRYPOINT = "quant_strategy_plugins.strategy_plugin_runner:run_market_regime_control_plugin" +PRODUCER_REPOSITORY = "QuantStrategyLab/QuantStrategyPlugins" +STRATEGY = "tqqq_growth_income" +PLUGIN = "market_regime_control" +MODE = "shadow" +_COMMIT_HEX = frozenset("0123456789abcdef") +_FORBIDDEN_KEY_PARTS = ("token", "secret", "password", "cookie", "jwt", "api_key") +_PAYLOAD_KEYS = { + "as_of", + "audit_summary", + "arbiter", + "canonical_route", + "component_signals", + "configured_mode", + "consumption_policy", + "effective_mode", + "execution_controls", + "generated_at", + "localized_messages", + "log_record", + "mode", + "notification", + "plugin", + "position_control", + "profile", + "schema_version", + "strategy", + "strategy_policy", + "suggested_action", + "target_type", + "would_trade_if_enabled", +} + + +class PresentPackageError(RuntimeError): + def __init__(self, code: str, exit_code: int) -> None: + super().__init__(code) + self.code = code + self.exit_code = exit_code + + +@dataclass(frozen=True) +class PublishedPackage: + path: Path + sha256: str + + +def _fail(code: str, exit_code: int) -> None: + raise PresentPackageError(code, exit_code) + + +def _sha256(value: bytes) -> str: + return hashlib.sha256(value).hexdigest() + + +def _canonical_json(value: Any) -> bytes: + return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False).encode("utf-8") + + +def _strict_json(raw: bytes, *, code: str, exit_code: int) -> Any: + def reject_duplicate_keys(pairs: list[tuple[str, Any]]) -> dict[str, Any]: + result: dict[str, Any] = {} + for key, value in pairs: + if key in result: + _fail(code, exit_code) + result[key] = value + return result + + try: + text = raw.decode("utf-8") + return json.loads( + text, + object_pairs_hook=reject_duplicate_keys, + parse_constant=lambda _value: _fail(code, exit_code), + ) + except (UnicodeDecodeError, TypeError, ValueError, json.JSONDecodeError): + _fail(code, exit_code) + + +def _is_sha256(value: object) -> bool: + return isinstance(value, str) and len(value) == 64 and set(value) <= _COMMIT_HEX + + +def _is_commit_sha(value: object) -> bool: + return isinstance(value, str) and len(value) == 40 and set(value) <= _COMMIT_HEX + + +def _is_int(value: object) -> bool: + return isinstance(value, int) and not isinstance(value, bool) + + +def _require_mapping(value: object, *, code: str, exit_code: int) -> dict[str, Any]: + if not isinstance(value, dict) or any(not isinstance(key, str) for key in value): + _fail(code, exit_code) + return value + + +def _require_exact_keys(value: object, keys: set[str], *, code: str, exit_code: int) -> dict[str, Any]: + mapping = _require_mapping(value, code=code, exit_code=exit_code) + if set(mapping) != keys: + _fail(code, exit_code) + return mapping + + +def _validate_json_value(value: Any, *, code: str, exit_code: int) -> None: + if value is None or isinstance(value, (str, bool)): + return + if _is_int(value): + return + if isinstance(value, float): + if math.isfinite(value): + return + _fail(code, exit_code) + if isinstance(value, list): + for child in value: + _validate_json_value(child, code=code, exit_code=exit_code) + return + if isinstance(value, dict): + for key, child in value.items(): + if not isinstance(key, str): + _fail(code, exit_code) + _validate_json_value(child, code=code, exit_code=exit_code) + return + _fail(code, exit_code) + + +def _reject_sensitive_keys(value: Any, *, code: str, exit_code: int) -> None: + if isinstance(value, dict): + for key, child in value.items(): + lowered = key.lower() + if lowered.startswith("ai_audit_") or any(part in lowered for part in _FORBIDDEN_KEY_PARTS): + _fail(code, exit_code) + _reject_sensitive_keys(child, code=code, exit_code=exit_code) + elif isinstance(value, list): + for child in value: + _reject_sensitive_keys(child, code=code, exit_code=exit_code) + + +def _git(*args: str) -> str: + return subprocess.check_output(["git", *args], text=True, stderr=subprocess.DEVNULL).strip() + + +def _only_private_stage_is_dirty(root: Path, status: str, private_stage: Path) -> bool: + try: + relative_stage = private_stage.resolve().relative_to(root).as_posix() + except ValueError: + return False + if not private_stage.name.startswith(".tqqq-market-regime-control-present-"): + return False + for record in filter(None, status.split("\0")): + if len(record) < 4 or record[:2] != "??": + return False + path = record[3:] + if path != relative_stage and not path.startswith(f"{relative_stage}/"): + return False + return bool(status) + + +def _verify_source_identity(expected_qsp_commit_sha: str, *, private_stage: Path | None = None) -> None: + if not _is_commit_sha(expected_qsp_commit_sha): + _fail("T2B2_PRODUCER_IDENTITY_INVALID", 2) + try: + root = Path(_git("rev-parse", "--show-toplevel")).resolve() + if Path.cwd().resolve() != root: + _fail("T2B2_PRODUCER_IDENTITY_INVALID", 2) + module_path = Path(__file__).resolve() + relative_module = module_path.relative_to(root) + if _git("ls-files", "--error-unmatch", str(relative_module)) != str(relative_module): + _fail("T2B2_PRODUCER_IDENTITY_INVALID", 2) + status = _git("status", "--porcelain=v1", "-z", "--untracked-files=all") + if status and (private_stage is None or not _only_private_stage_is_dirty(root, status, private_stage)): + _fail("T2B2_PRODUCER_IDENTITY_INVALID", 2) + if _git("rev-parse", "HEAD") != expected_qsp_commit_sha: + _fail("T2B2_PRODUCER_IDENTITY_INVALID", 2) + except (OSError, subprocess.SubprocessError, ValueError): + _fail("T2B2_PRODUCER_IDENTITY_INVALID", 2) + + +def _require_iso_date(value: object, *, code: str, exit_code: int) -> str: + if not isinstance(value, str): + _fail(code, exit_code) + try: + parsed = date.fromisoformat(value) + except ValueError: + _fail(code, exit_code) + if parsed.isoformat() != value: + _fail(code, exit_code) + return value + + +def _resolve_input_path(config_path: Path, value: object, *, code: str, exit_code: int) -> Path: + if not isinstance(value, str) or not value.strip(): + _fail(code, exit_code) + path = Path(value) + if not path.is_absolute(): + path = config_path.parent / path + return path + + +def _read_regular_csv(path: Path, *, code: str, exit_code: int) -> bytes: + try: + if path.is_symlink() or not path.is_file() or path.suffix.lower() != ".csv": + _fail(code, exit_code) + return path.read_bytes() + except OSError: + _fail(code, exit_code) + + +def _copy_staged_csv(source: Path, destination: Path) -> tuple[bytes, dict[str, Any]]: + raw = _read_regular_csv(source, code="T2B2_PRODUCER_INPUT_INVALID", exit_code=2) + destination.parent.mkdir(parents=True, exist_ok=True) + destination.write_bytes(raw) + return raw, {"format": "csv", "sha256": _sha256(raw), "size_bytes": len(raw)} + + +def _select_target(config_path: Path, as_of: str, session_id: str) -> tuple[dict[str, Any], str]: + try: + config = load_plugin_config(config_path) + entries = config.get("strategy_plugins") + if not isinstance(entries, list): + _fail("T2B2_PRODUCER_INPUT_INVALID", 2) + matches = [] + for entry in entries: + if not isinstance(entry, Mapping): + _fail("T2B2_PRODUCER_INPUT_INVALID", 2) + flattened = _flatten_strategy_plugin_entry(entry) + if flattened.get("strategy") == STRATEGY and flattened.get("plugin") == PLUGIN: + matches.append(flattened) + if len(matches) != 1: + _fail("T2B2_PRODUCER_INPUT_INVALID", 2) + target = matches[0] + if target.get("enabled") is not True: + _fail("T2B2_PRODUCER_INPUT_INVALID", 2) + mode = target.get("mode", config.get("default_mode")) + if mode != MODE: + _fail("T2B2_PRODUCER_INPUT_INVALID", 2) + if target.get("as_of") != as_of or session_id != f"XNAS:{as_of}": + _fail("T2B2_PRODUCER_INPUT_INVALID", 2) + _reject_sensitive_keys(target, code="T2B2_PRODUCER_INPUT_INVALID", exit_code=2) + _validate_json_value(target, code="T2B2_PRODUCER_INPUT_INVALID", exit_code=2) + return target, MODE + except PresentPackageError: + raise + except (OSError, TypeError, ValueError): + _fail("T2B2_PRODUCER_INPUT_INVALID", 2) + + +def _validate_payload(raw: bytes, *, as_of: str, code: str, exit_code: int) -> None: + payload = _require_exact_keys(_strict_json(raw, code=code, exit_code=exit_code), _PAYLOAD_KEYS, code=code, exit_code=exit_code) + _validate_json_value(payload, code=code, exit_code=exit_code) + if ( + payload["schema_version"] != "market_regime_control.v1" + or payload["profile"] != "market_regime_control" + or payload["strategy"] != STRATEGY + or payload["plugin"] != PLUGIN + or payload["target_type"] != "strategy" + or payload["mode"] != MODE + or payload["configured_mode"] != MODE + or payload["effective_mode"] != MODE + or payload["as_of"] != as_of + ): + _fail(code, exit_code) + consumption_policy = _require_mapping(payload["consumption_policy"], code=code, exit_code=exit_code) + if ( + consumption_policy.get("plugin") != PLUGIN + or consumption_policy.get("strategy") != STRATEGY + or consumption_policy.get("position_control_allowed") is not True + or consumption_policy.get("evidence_status") != "automation_approved" + ): + _fail(code, exit_code) + + +def _verify_prices_identity(value: object, *, code: str, exit_code: int) -> None: + mapping = _require_mapping(value, code=code, exit_code=exit_code) + if set(mapping) != {"format", "sha256", "size_bytes"}: + _fail(code, exit_code) + if mapping.get("format") != "csv" or not _is_sha256(mapping.get("sha256")): + _fail(code, exit_code) + if not _is_int(mapping.get("size_bytes")) or mapping["size_bytes"] < 0: + _fail(code, exit_code) + + +def _verify_external_context_identity(value: object, *, code: str, exit_code: int) -> None: + mapping = _require_mapping(value, code=code, exit_code=exit_code) + if mapping.get("status") == "ABSENT": + if set(mapping) != {"status"}: + _fail(code, exit_code) + return + if set(mapping) != {"status", "format", "sha256", "size_bytes"}: + _fail(code, exit_code) + if mapping.get("status") != "PRESENT" or mapping.get("format") != "csv" or not _is_sha256(mapping.get("sha256")): + _fail(code, exit_code) + if not _is_int(mapping.get("size_bytes")) or mapping["size_bytes"] < 0: + _fail(code, exit_code) + + +def _verify_present_package(raw: bytes, *, expected_qsp_commit_sha: str | None = None) -> dict[str, Any]: + code = "T2B2_PRODUCER_PUBLISH_FAILED" + package = _require_exact_keys( + _strict_json(raw, code=code, exit_code=4), + {"as_of", "config", "inputs", "payload", "producer", "schema", "session_id", "status", "subject"}, + code=code, + exit_code=4, + ) + _validate_json_value(package, code=code, exit_code=4) + if raw != _canonical_json(package): + _fail(code, 4) + as_of = _require_iso_date(package["as_of"], code=code, exit_code=4) + if package["schema"] != SCHEMA or package["status"] != "PRESENT" or package["session_id"] != f"XNAS:{as_of}": + _fail(code, 4) + subject = _require_exact_keys(package["subject"], {"mode", "plugin", "strategy"}, code=code, exit_code=4) + if subject != {"mode": MODE, "plugin": PLUGIN, "strategy": STRATEGY}: + _fail(code, 4) + producer = _require_exact_keys(package["producer"], {"commit_sha", "entrypoint", "repository"}, code=code, exit_code=4) + if ( + not _is_commit_sha(producer["commit_sha"]) + or producer["entrypoint"] != PRODUCER_ENTRYPOINT + or producer["repository"] != PRODUCER_REPOSITORY + or (expected_qsp_commit_sha is not None and producer["commit_sha"] != expected_qsp_commit_sha) + ): + _fail(code, 4) + config = _require_exact_keys(package["config"], {"sha256", "value"}, code=code, exit_code=4) + value = _require_mapping(config["value"], code=code, exit_code=4) + _reject_sensitive_keys(value, code=code, exit_code=4) + if ( + not _is_sha256(config["sha256"]) + or config["sha256"] != _sha256(_canonical_json(value)) + or value.get("strategy") != STRATEGY + or value.get("plugin") != PLUGIN + or value.get("enabled") is not True + or value.get("mode") != MODE + or value.get("as_of") != as_of + or value.get("prices") != "@input:prices" + ): + _fail(code, 4) + inputs = _require_exact_keys(package["inputs"], {"external_context", "prices"}, code=code, exit_code=4) + _verify_prices_identity(inputs["prices"], code=code, exit_code=4) + _verify_external_context_identity(inputs["external_context"], code=code, exit_code=4) + payload = _require_exact_keys(package["payload"], {"bytes_b64", "schema_version", "sha256", "size_bytes"}, code=code, exit_code=4) + if ( + not isinstance(payload["bytes_b64"], str) + or payload["schema_version"] != "market_regime_control.v1" + or not _is_sha256(payload["sha256"]) + or not _is_int(payload["size_bytes"]) + or payload["size_bytes"] < 0 + ): + _fail(code, 4) + try: + payload_bytes = base64.b64decode(payload["bytes_b64"], validate=True) + except (ValueError, TypeError): + _fail(code, 4) + if _sha256(payload_bytes) != payload["sha256"] or len(payload_bytes) != payload["size_bytes"]: + _fail(code, 4) + _validate_payload(payload_bytes, as_of=as_of, code=code, exit_code=4) + return package + + +def _readback_package(path: Path, *, expected_qsp_commit_sha: str) -> None: + try: + if path.is_symlink() or not path.is_file(): + _fail("T2B2_PRODUCER_PUBLISH_FAILED", 4) + _verify_present_package(path.read_bytes(), expected_qsp_commit_sha=expected_qsp_commit_sha) + except OSError: + _fail("T2B2_PRODUCER_PUBLISH_FAILED", 4) + + +def capture_tqqq_market_regime_control_present( + *, + config_path: str | Path, + output_dir: str | Path, + expected_qsp_commit_sha: str, + as_of: str, + session_id: str, +) -> PublishedPackage: + as_of = _require_iso_date(as_of, code="T2B2_PRODUCER_INPUT_INVALID", exit_code=2) + _verify_source_identity(expected_qsp_commit_sha) + config_path = Path(config_path) + if config_path.is_symlink() or not config_path.is_file(): + _fail("T2B2_PRODUCER_INPUT_INVALID", 2) + config_path = config_path.resolve() + target, default_mode = _select_target(config_path, as_of, session_id) + prices_source = _resolve_input_path(config_path, target.get("prices"), code="T2B2_PRODUCER_INPUT_INVALID", exit_code=2) + external_source = ( + _resolve_input_path(config_path, target["external_context"], code="T2B2_PRODUCER_INPUT_INVALID", exit_code=2) + if "external_context" in target + else None + ) + destination_root = Path(output_dir).resolve() + try: + destination_root.mkdir(parents=True, exist_ok=True) + except OSError: + _fail("T2B2_PRODUCER_PUBLISH_FAILED", 4) + + with tempfile.TemporaryDirectory(prefix=".tqqq-market-regime-control-present-", dir=destination_root) as temporary: + stage = Path(temporary) + staged_prices, prices_identity = _copy_staged_csv(prices_source, stage / "inputs" / "prices.csv") + external_identity: dict[str, Any] = {"status": "ABSENT"} + staged_external: bytes | None = None + if external_source is not None: + staged_external, external_metadata = _copy_staged_csv(external_source, stage / "inputs" / "external_context.csv") + external_identity = {"status": "PRESENT", **external_metadata} + staged_config = dict(target) + staged_config["prices"] = str(stage / "inputs" / "prices.csv") + if staged_external is not None: + staged_config["external_context"] = str(stage / "inputs" / "external_context.csv") + staged_config["output_dir"] = str(stage / "producer-output") + try: + run_market_regime_control_plugin(staged_config, default_mode) + except PresentPackageError: + raise + except Exception: + _fail("T2B2_PRODUCER_ARTIFACT_INVALID", 3) + if _read_regular_csv(stage / "inputs" / "prices.csv", code="T2B2_PRODUCER_INPUT_INVALID", exit_code=2) != staged_prices: + _fail("T2B2_PRODUCER_INPUT_INVALID", 2) + if staged_external is not None and ( + _read_regular_csv(stage / "inputs" / "external_context.csv", code="T2B2_PRODUCER_INPUT_INVALID", exit_code=2) + != staged_external + ): + _fail("T2B2_PRODUCER_INPUT_INVALID", 2) + try: + producer_output = stage / "producer-output" + dated_payload = _read_regular_file(producer_output / "signals" / f"{as_of}.json") + latest_payload = _read_regular_file(producer_output / "latest_signal.json") + if dated_payload != latest_payload: + _fail("T2B2_PRODUCER_ARTIFACT_INVALID", 3) + _validate_payload(dated_payload, as_of=as_of, code="T2B2_PRODUCER_ARTIFACT_INVALID", exit_code=3) + except PresentPackageError: + raise + except OSError: + _fail("T2B2_PRODUCER_ARTIFACT_INVALID", 3) + config_value = dict(target) + config_value.pop("output_dir", None) + config_value["mode"] = MODE + config_value["prices"] = "@input:prices" + if "external_context" in config_value: + config_value["external_context"] = "@input:external_context" + package = { + "as_of": as_of, + "config": {"sha256": _sha256(_canonical_json(config_value)), "value": config_value}, + "inputs": {"external_context": external_identity, "prices": prices_identity}, + "payload": { + "bytes_b64": base64.b64encode(dated_payload).decode("ascii"), + "schema_version": "market_regime_control.v1", + "sha256": _sha256(dated_payload), + "size_bytes": len(dated_payload), + }, + "producer": { + "commit_sha": expected_qsp_commit_sha, + "entrypoint": PRODUCER_ENTRYPOINT, + "repository": PRODUCER_REPOSITORY, + }, + "schema": SCHEMA, + "session_id": session_id, + "status": "PRESENT", + "subject": {"mode": MODE, "plugin": PLUGIN, "strategy": STRATEGY}, + } + package_bytes = _canonical_json(package) + staged_package = stage / "package.json" + try: + staged_package.write_bytes(package_bytes) + _readback_package(staged_package, expected_qsp_commit_sha=expected_qsp_commit_sha) + _readback_package(staged_package, expected_qsp_commit_sha=expected_qsp_commit_sha) + _verify_source_identity(expected_qsp_commit_sha, private_stage=stage) + shutil.rmtree(stage / "producer-output") + if (stage / "producer-output").exists(): + _fail("T2B2_PRODUCER_PUBLISH_FAILED", 4) + digest = _sha256(package_bytes) + destination = destination_root / f"tqqq-market-regime-control-present-{as_of}-{digest}.json" + if destination.exists() or destination.is_symlink(): + _fail("T2B2_PRODUCER_PUBLISH_FAILED", 4) + os.replace(staged_package, destination) + except PresentPackageError: + raise + except OSError: + _fail("T2B2_PRODUCER_PUBLISH_FAILED", 4) + return PublishedPackage(path=destination, sha256=digest) + + +def _read_regular_file(path: Path) -> bytes: + if path.is_symlink() or not path.is_file(): + _fail("T2B2_PRODUCER_ARTIFACT_INVALID", 3) + return path.read_bytes() + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description="Capture one TQQQ market-regime-control PRESENT package.") + parser.add_argument("--config", required=True) + parser.add_argument("--output-dir", required=True) + parser.add_argument("--expected-qsp-commit-sha", required=True) + parser.add_argument("--as-of", required=True) + parser.add_argument("--session-id", required=True) + return parser + + +def main(argv: list[str] | None = None) -> int: + args = build_parser().parse_args(argv) + try: + package = capture_tqqq_market_regime_control_present( + config_path=args.config, + output_dir=args.output_dir, + expected_qsp_commit_sha=args.expected_qsp_commit_sha, + as_of=args.as_of, + session_id=args.session_id, + ) + except PresentPackageError as error: + print(f"ERROR {error.code}", file=sys.stderr) + return error.exit_code + except Exception: + print("ERROR T2B2_PRODUCER_PUBLISH_FAILED", file=sys.stderr) + return 70 + print(f"{package.path} {package.sha256}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_tqqq_market_regime_control_present.py b/tests/test_tqqq_market_regime_control_present.py new file mode 100644 index 0000000..1b9fc17 --- /dev/null +++ b/tests/test_tqqq_market_regime_control_present.py @@ -0,0 +1,263 @@ +from __future__ import annotations + +import base64 +import hashlib +import json +import shutil +import subprocess +from pathlib import Path + +import pandas as pd +import pytest + +import quant_strategy_plugins.tqqq_market_regime_control_present as present_module +from quant_strategy_plugins.strategy_plugin_runner import run_market_regime_control_plugin + + +AS_OF = "2025-12-31" +EXPECTED_COMMIT = subprocess.check_output( + ["git", "rev-parse", "HEAD"], text=True, cwd=Path(__file__).resolve().parents[1] +).strip() + + +def _market_regime_prices() -> pd.DataFrame: + dates = pd.bdate_range("2025-01-02", periods=260) + qqq = pd.Series([100.0 + index * 0.10 for index in range(len(dates))], index=dates) + qqq.iloc[-30:] = pd.Series([125.0 - index * (45.0 / 29.0) for index in range(30)], index=dates[-30:]) + vix = pd.Series(15.0, index=dates) + vix.iloc[-6:] = [24.0, 27.0, 31.0, 34.0, 38.0, 41.0] + hyg = pd.Series(100.0, index=dates) + hyg.iloc[-22:] = pd.Series([100.0 - index * (9.0 / 21.0) for index in range(22)], index=dates[-22:]) + ief = pd.Series(100.0, index=dates) + ief.iloc[-22:] = pd.Series([100.0 + index * (3.0 / 21.0) for index in range(22)], index=dates[-22:]) + rows: list[dict[str, object]] = [] + for symbol, series in {"QQQ": qqq, "TQQQ": qqq * 3.0, "VIX": vix, "HYG": hyg, "IEF": ief}.items(): + for as_of, close in series.items(): + rows.append({"symbol": symbol, "as_of": as_of, "close": close, "volume": 1_000_000}) + return pd.DataFrame(rows) + + +def _config(tmp_path: Path) -> Path: + prices_path = tmp_path / "prices.csv" + _market_regime_prices().to_csv(prices_path, index=False) + config_path = tmp_path / "strategy_plugins.toml" + config_path.write_text( + f""" +default_mode = "shadow" + +[[strategy_plugins]] +strategy = "tqqq_growth_income" +plugin = "market_regime_control" +enabled = true +mode = "shadow" + +[strategy_plugins.inputs] +prices = "{prices_path}" +as_of = "{AS_OF}" +vix_symbols = ["VIX"] +credit_pairs = ["HYG:IEF"] +crisis_enabled = false +taco_enabled = false +crisis_score_threshold = 99.0 + +[strategy_plugins.outputs] +output_dir = "{tmp_path / 'writer-output'}" +""".strip(), + encoding="utf-8", + ) + return config_path + + +def _capture(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, *, output_dir: Path | None = None): + source_checks: list[Path | None] = [] + monkeypatch.setattr( + present_module, + "_verify_source_identity", + lambda _expected, private_stage=None: source_checks.append(private_stage), + ) + calls: list[dict[str, object]] = [] + + def run_once(config: dict[str, object], mode: str): + calls.append(dict(config)) + return run_market_regime_control_plugin(config, mode) + + monkeypatch.setattr(present_module, "run_market_regime_control_plugin", run_once) + package = present_module.capture_tqqq_market_regime_control_present( + config_path=_config(tmp_path), + output_dir=output_dir or tmp_path / "packages", + expected_qsp_commit_sha=EXPECTED_COMMIT, + as_of=AS_OF, + session_id=f"XNAS:{AS_OF}", + ) + return package, calls, source_checks + + +def test_capture_publishes_one_canonical_package_after_one_current_producer_call(tmp_path, monkeypatch) -> None: + package, calls, source_checks = _capture(tmp_path, monkeypatch) + + package_bytes = package.path.read_bytes() + decoded = json.loads(package_bytes) + payload_bytes = base64.b64decode(decoded["payload"]["bytes_b64"], validate=True) + + assert len(calls) == 1 + assert source_checks[0] is None + assert source_checks[1] is not None + assert package.path.parent == tmp_path / "packages" + assert package.path.name == f"tqqq-market-regime-control-present-{AS_OF}-{package.sha256}.json" + assert package.sha256 == hashlib.sha256(package_bytes).hexdigest() + assert package_bytes == json.dumps( + decoded, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False + ).encode("utf-8") + assert decoded["schema"] == "qsl.tqqq_market_regime_control_present.v1" + assert decoded["session_id"] == f"XNAS:{AS_OF}" + assert set(decoded["inputs"]["prices"]) == {"format", "sha256", "size_bytes"} + assert decoded["inputs"]["prices"]["format"] == "csv" + assert decoded["inputs"]["external_context"] == {"status": "ABSENT"} + assert "output_dir" not in decoded["config"]["value"] + assert decoded["config"]["value"]["mode"] == "shadow" + assert decoded["payload"]["sha256"] == hashlib.sha256(payload_bytes).hexdigest() + assert list((tmp_path / "packages").iterdir()) == [package.path] + + +def test_capture_allows_unignored_in_repository_output_and_removes_private_stage(tmp_path, monkeypatch) -> None: + output_dir = Path.cwd() / f"tqqq-present-output-{tmp_path.name}" + try: + package, _, source_checks = _capture(tmp_path, monkeypatch, output_dir=output_dir) + + assert package.path.parent == output_dir + assert source_checks[0] is None + assert source_checks[1] is not None + assert source_checks[1].parent == output_dir + assert not list(output_dir.glob(".tqqq-market-regime-control-present-*")) + finally: + shutil.rmtree(output_dir, ignore_errors=True) + + +def test_capture_rejects_staged_input_mutation_without_a_final_package(tmp_path, monkeypatch) -> None: + monkeypatch.setattr(present_module, "_verify_source_identity", lambda _expected: None) + + def mutate_after_run(config: dict[str, object], mode: str): + result = run_market_regime_control_plugin(config, mode) + Path(str(config["prices"])).write_bytes(b"changed") + return result + + monkeypatch.setattr(present_module, "run_market_regime_control_plugin", mutate_after_run) + destination = tmp_path / "packages" + + with pytest.raises(present_module.PresentPackageError, match="T2B2_PRODUCER_INPUT_INVALID"): + present_module.capture_tqqq_market_regime_control_present( + config_path=_config(tmp_path), + output_dir=destination, + expected_qsp_commit_sha=EXPECTED_COMMIT, + as_of=AS_OF, + session_id=f"XNAS:{AS_OF}", + ) + + assert not list(destination.glob("*.json")) + + +def test_capture_rejects_dated_latest_mismatch_without_a_final_package(tmp_path, monkeypatch) -> None: + monkeypatch.setattr(present_module, "_verify_source_identity", lambda _expected: None) + + def corrupt_latest(config: dict[str, object], mode: str): + result = run_market_regime_control_plugin(config, mode) + Path(str(config["output_dir"])).joinpath("latest_signal.json").write_text("{}", encoding="utf-8") + return result + + monkeypatch.setattr(present_module, "run_market_regime_control_plugin", corrupt_latest) + destination = tmp_path / "packages" + + with pytest.raises(present_module.PresentPackageError, match="T2B2_PRODUCER_ARTIFACT_INVALID"): + present_module.capture_tqqq_market_regime_control_present( + config_path=_config(tmp_path), + output_dir=destination, + expected_qsp_commit_sha=EXPECTED_COMMIT, + as_of=AS_OF, + session_id=f"XNAS:{AS_OF}", + ) + + assert not list(destination.glob("*.json")) + + +def test_source_identity_rejects_a_wrong_expected_commit(monkeypatch) -> None: + monkeypatch.setattr(present_module, "_git", lambda *_args: str(Path.cwd())) + + with pytest.raises(present_module.PresentPackageError, match="T2B2_PRODUCER_IDENTITY_INVALID"): + present_module._verify_source_identity("0" * 40) + + +def _source_identity_git(monkeypatch, status: str) -> tuple[Path, list[tuple[str, ...]]]: + root = Path.cwd().resolve() + relative_module = Path(present_module.__file__).resolve().relative_to(root) + calls: list[tuple[str, ...]] = [] + + def fake_git(*args: str) -> str: + calls.append(args) + if args == ("rev-parse", "--show-toplevel"): + return str(root) + if args == ("ls-files", "--error-unmatch", str(relative_module)): + return str(relative_module) + if args and args[0] == "status": + return status + if args == ("rev-parse", "HEAD"): + return EXPECTED_COMMIT + raise AssertionError(args) + + monkeypatch.setattr(present_module, "_git", fake_git) + return root, calls + + +def test_second_source_identity_check_allows_only_its_private_stage(monkeypatch) -> None: + root = Path.cwd().resolve() + private_stage = root / "unignored-output" / ".tqqq-market-regime-control-present-test" + relative_stage = private_stage.relative_to(root).as_posix() + _, calls = _source_identity_git(monkeypatch, f"?? {relative_stage}/\0") + + present_module._verify_source_identity(EXPECTED_COMMIT, private_stage=private_stage) + + assert ("status", "--porcelain=v1", "-z", "--untracked-files=all") in calls + + with pytest.raises(present_module.PresentPackageError, match="T2B2_PRODUCER_IDENTITY_INVALID"): + present_module._verify_source_identity(EXPECTED_COMMIT) + + +@pytest.mark.parametrize("status", [" M src/unrelated.py\0", "?? unrelated.txt\0"]) +def test_second_source_identity_check_rejects_unrelated_tracked_or_untracked_dirt(monkeypatch, status: str) -> None: + root = Path.cwd().resolve() + private_stage = root / "unignored-output" / ".tqqq-market-regime-control-present-test" + _source_identity_git(monkeypatch, f"?? {private_stage.relative_to(root).as_posix()}/\0{status}") + + with pytest.raises(present_module.PresentPackageError, match="T2B2_PRODUCER_IDENTITY_INVALID"): + present_module._verify_source_identity(EXPECTED_COMMIT, private_stage=private_stage) + + +def test_identity_failure_does_not_create_a_final_package(tmp_path, monkeypatch) -> None: + monkeypatch.setattr( + present_module, + "_verify_source_identity", + lambda *_args, **_kwargs: (_ for _ in ()).throw(present_module.PresentPackageError("T2B2_PRODUCER_IDENTITY_INVALID", 2)), + ) + destination = tmp_path / "packages" + + with pytest.raises(present_module.PresentPackageError, match="T2B2_PRODUCER_IDENTITY_INVALID"): + present_module.capture_tqqq_market_regime_control_present( + config_path=_config(tmp_path), + output_dir=destination, + expected_qsp_commit_sha=EXPECTED_COMMIT, + as_of=AS_OF, + session_id=f"XNAS:{AS_OF}", + ) + + assert not destination.exists() + + +def test_strict_readback_rejects_an_unknown_package_field(tmp_path, monkeypatch) -> None: + package, _, _ = _capture(tmp_path, monkeypatch) + decoded = json.loads(package.path.read_bytes()) + decoded["unknown"] = "rejected" + + with pytest.raises(present_module.PresentPackageError, match="T2B2_PRODUCER_PUBLISH_FAILED"): + present_module._verify_present_package( + json.dumps(decoded, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8"), + expected_qsp_commit_sha=EXPECTED_COMMIT, + )