Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
52 changes: 52 additions & 0 deletions .github/workflows/drift-check.yml
Original file line number Diff line number Diff line change
@@ -0,0 +1,52 @@
name: Drift Check

# Daily drift detection for crypto strategies.

on:
schedule:
- cron: "0 6 * * *"
workflow_dispatch:

permissions:
contents: read
issues: write

jobs:
drift:
runs-on: ubuntu-latest
timeout-minutes: 15
env:
FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
STRATEGY_DOMAIN: crypto

steps:
- name: Checkout
uses: actions/checkout@v6

- name: Checkout QuantPlatformKit
uses: actions/checkout@v6
with:
repository: QuantStrategyLab/QuantPlatformKit
ref: main
path: external/QuantPlatformKit

- name: Set up Python
uses: actions/setup-python@v6
with:
python-version: "3.11"

- name: Install dependencies
run: |
set -euo pipefail
python -m pip install --upgrade pip
python -m pip install -e . pandas
python -m pip install --no-deps -e external/QuantPlatformKit

- name: Run drift detection
run: quant-lifecycle drift --domain crypto --no-alerts

- name: Sync drift alerts to GitHub Issues
env:
GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
GITHUB_REPOSITORY: ${{ github.repository }}
run: python scripts/run_drift_github_issues.py
76 changes: 76 additions & 0 deletions .github/workflows/evidence-gate.yml
Original file line number Diff line number Diff line change
@@ -0,0 +1,76 @@
name: Evidence Gate

# Block PRs that promote catalog status without a valid evidence package.

on:
pull_request:
types: [opened, synchronize, reopened, ready_for_review]
paths:
- "src/**/catalog.py"
- "src/**/combo_manifests.py"
- "docs/evidence/**"
- "evidence/**"

permissions:
contents: read
pull-requests: read

concurrency:
group: evidence-gate-${{ github.event.pull_request.number }}
cancel-in-progress: true

jobs:
gate:
if: github.event.pull_request.draft == false
runs-on: ubuntu-latest
timeout-minutes: 10
env:
FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"

steps:
- name: Checkout
uses: actions/checkout@v6
with:
fetch-depth: 0

- name: Resolve QuantPlatformKit ref
id: quant-platform-kit-ref
run: |
set -euo pipefail
ref="main"
if [ -n "${GITHUB_HEAD_REF:-}" ]; then
case "${GITHUB_HEAD_REF}" in
dependabot/*)
;;
*)
if git ls-remote --exit-code --heads https://github.com/QuantStrategyLab/QuantPlatformKit.git "${GITHUB_HEAD_REF}" >/dev/null 2>&1; then
ref="${GITHUB_HEAD_REF}"
fi
;;
esac
fi
echo "ref=${ref}" >> "$GITHUB_OUTPUT"

- name: Checkout QuantPlatformKit
uses: actions/checkout@v6
with:
repository: QuantStrategyLab/QuantPlatformKit
ref: ${{ steps.quant-platform-kit-ref.outputs.ref }}
path: external/QuantPlatformKit

- name: Set up Python
uses: actions/setup-python@v6
with:
python-version: "3.11"

- name: Install dependencies
run: |
set -euo pipefail
python -m pip install --upgrade pip
python -m pip install -e . pandas
python -m pip install --no-deps -e external/QuantPlatformKit

- name: Evaluate Evidence Gate
env:
GITHUB_BASE_REF: ${{ github.base_ref }}
run: python scripts/gate_evidence_package.py
129 changes: 129 additions & 0 deletions scripts/gate_evidence_package.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,129 @@
#!/usr/bin/env python3
"""PR gate: require a valid evidence package when catalog status is promoted."""

from __future__ import annotations

import os
import re
import subprocess
import sys
from pathlib import Path

GATE_STAGES = frozenset(
{
"ai_monitored_candidate",
"shadow_candidate",
"live_candidate",
"runtime_enabled",
}
)
STATUS_ADDED_RE = re.compile(r'^\+.*status="([^"]+)"')
EVIDENCE_SUFFIXES = {".json", ".toml"}


def _git_diff(base_ref: str) -> str:
result = subprocess.run(
["git", "diff", f"origin/{base_ref}...HEAD", "--", "src"],
capture_output=True,
text=True,
check=False,
)
if result.returncode != 0:
result = subprocess.run(
["git", "diff", f"{base_ref}...HEAD", "--", "src"],
capture_output=True,
text=True,
check=True,
)
return result.stdout


def _promotion_detected(diff: str) -> bool:
if "status=" not in diff:
return False
return any(match.group(1) in GATE_STAGES for line in diff.splitlines() if (match := STATUS_ADDED_RE.match(line)))


def _evidence_paths_from_diff(diff: str) -> list[Path]:
paths: list[Path] = []
for line in diff.splitlines():
if not line.startswith("+++ b/"):
continue
candidate = Path(line[6:])
if candidate.suffix.lower() not in EVIDENCE_SUFFIXES:
continue
if "evidence" in candidate.parts or candidate.parent.name == "evidence":
paths.append(candidate)
return paths


def _discover_evidence_files(diff: str) -> list[Path]:
discovered = _evidence_paths_from_diff(diff)
for folder in (Path("docs/evidence"), Path("evidence")):
if folder.is_dir():
discovered.extend(path for path in folder.iterdir() if path.suffix.lower() in EVIDENCE_SUFFIXES)
explicit = os.environ.get("EVIDENCE_PACKAGE_PATH", "").strip()
if explicit:
discovered.append(Path(explicit))
return sorted({path for path in discovered if path.exists()})


def _validate_with_lifecycle(path: Path) -> tuple[bool, list[str]]:
from quant_platform_kit.strategy_lifecycle.evidence_gate import validate_evidence_package_file

result = validate_evidence_package_file(path)
issues = list(result.issues)
return result.valid, issues


def _validate_with_promotion_standard(path: Path) -> tuple[bool, list[str]]:
script = Path("external/QuantPlatformKit/scripts/validate_strategy_evidence_package.py")
if not script.exists():
return True, []
result = subprocess.run(
[sys.executable, str(script), str(path)],
capture_output=True,
text=True,
check=False,
)
if result.returncode == 0:
return True, []
issues = [line for line in result.stderr.splitlines() if line.strip()]
issues.extend(line for line in result.stdout.splitlines() if line.strip())
return False, issues or ["promotion evidence package validation failed"]


def main() -> int:
base_ref = os.environ.get("GITHUB_BASE_REF", "main").strip() or "main"
diff = _git_diff(base_ref)

if not _promotion_detected(diff):
print("[evidence-gate] No lifecycle status promotion detected; skipping validation")
return 0

evidence_files = _discover_evidence_files(diff)
if not evidence_files:
print(
"::error::Catalog status promotion detected but no evidence package file was found. "
"Add docs/evidence/<profile>.json with the 11 required artifacts.",
file=sys.stderr,
)
return 1

failed = False
for path in evidence_files:
lifecycle_ok, lifecycle_issues = _validate_with_lifecycle(path)
standard_ok, standard_issues = _validate_with_promotion_standard(path)
if lifecycle_ok and standard_ok:
print(f"[evidence-gate] PASS {path}")
continue
failed = True
print(f"[evidence-gate] FAIL {path}", file=sys.stderr)
for issue in lifecycle_issues + standard_issues:
print(f" - {issue}", file=sys.stderr)

return 1 if failed else 0


if __name__ == "__main__":
raise SystemExit(main())
42 changes: 42 additions & 0 deletions scripts/run_drift_github_issues.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,42 @@
#!/usr/bin/env python3
"""Create GitHub issues for drift alerts after quant-lifecycle drift detection."""

from __future__ import annotations

import os
import sys


def main() -> int:
domain = os.environ.get("STRATEGY_DOMAIN", "").strip()
if not domain:
print("::error::STRATEGY_DOMAIN is required", file=sys.stderr)
return 1

repository = os.environ.get("GITHUB_REPOSITORY", "").strip()
if "/" in repository:
owner, repo = repository.split("/", 1)
os.environ.setdefault("CODEX_AUDIT_ORG", owner)
os.environ.setdefault("CODEX_AUDIT_ORCHESTRATOR_REPO", repo)

from quant_platform_kit.strategy_lifecycle.codex_integration import create_issues_for_domain
from quant_platform_kit.strategy_lifecycle.drift_detector import run_drift_detection

drifts = run_drift_detection(domain)
critical = sum(1 for item in drifts if getattr(getattr(item, "status", None), "value", "") == "critical")
review = sum(1 for item in drifts if getattr(getattr(item, "status", None), "value", "") == "review")
print(f"[drift-check] domain={domain} checked={len(drifts)} review={review} critical={critical}")

results = create_issues_for_domain(domain, dry_run=False)
created = [item for item in results if item.get("issue_url")]
errors = [item for item in results if item.get("error")]
print(f"[drift-check] issues_created={len(created)} errors={len(errors)}")
for item in created:
print(f" - {item.get('issue_url')}")
for item in errors:
print(f"::warning::{item.get('title')}: {item.get('error')}", file=sys.stderr)
return 0


if __name__ == "__main__":
raise SystemExit(main())
Loading