diff --git a/.github/workflows/strategy_optimization_watcher.yml b/.github/workflows/strategy_optimization_watcher.yml new file mode 100644 index 00000000..4909886a --- /dev/null +++ b/.github/workflows/strategy_optimization_watcher.yml @@ -0,0 +1,155 @@ +name: Strategy Optimization Watcher + +on: + workflow_dispatch: + inputs: + source_repo: + description: "Repository that owns strategy metrics and receives optimization issues" + required: true + default: "QuantStrategyLab/CryptoLivePoolPipelines" + source_ref: + description: "Source repository ref to inspect" + required: false + default: "main" + metrics_path: + description: "JSON metrics payload path inside the source repository" + required: false + default: "data/output/strategy_metrics.json" + dry_run: + description: "Do not create GitHub issues" + required: false + type: boolean + default: true + schedule: + - cron: "17 3 * * *" + +permissions: + contents: read + issues: write + +concurrency: + group: strategy-optimization-watcher-${{ github.event.inputs.source_repo || vars.STRATEGY_WATCH_SOURCE_REPO || 'QuantStrategyLab/CryptoLivePoolPipelines' }} + cancel-in-progress: false + +jobs: + strategy-optimization-watcher: + runs-on: ubuntu-latest + timeout-minutes: 15 + env: + FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true" + SOURCE_REPO: ${{ github.event.inputs.source_repo || vars.STRATEGY_WATCH_SOURCE_REPO || 'QuantStrategyLab/CryptoLivePoolPipelines' }} + SOURCE_REF: ${{ github.event.inputs.source_ref || vars.STRATEGY_WATCH_SOURCE_REF || 'main' }} + METRICS_PATH: ${{ github.event.inputs.metrics_path || vars.STRATEGY_WATCH_METRICS_PATH || 'data/output/strategy_metrics.json' }} + STRATEGY_WATCH_DRY_RUN: ${{ github.event_name == 'workflow_dispatch' && format('{0}', inputs.dry_run) || vars.STRATEGY_WATCH_DRY_RUN || 'true' }} + ALLOWED_SOURCE_REPOS: ${{ vars.STRATEGY_WATCH_ALLOWED_SOURCE_REPOS || 'QuantStrategyLab/CryptoLivePoolPipelines' }} + ALLOWED_SOURCE_REFS: ${{ vars.STRATEGY_WATCH_ALLOWED_SOURCE_REFS || 'main' }} + steps: + - name: Checkout Bridge + uses: actions/checkout@v6.0.3 + with: + path: bridge + persist-credentials: false + + - name: Detect GitHub App Credentials + id: app_credentials + env: + APP_ID: ${{ vars.CROSS_REPO_GITHUB_APP_ID }} + APP_PRIVATE_KEY: ${{ secrets.CROSS_REPO_GITHUB_APP_PRIVATE_KEY }} + run: | + set -euo pipefail + if [ -n "${APP_ID:-}" ] && [ -n "${APP_PRIVATE_KEY:-}" ]; then + echo "available=true" >> "$GITHUB_OUTPUT" + else + echo "available=false" >> "$GITHUB_OUTPUT" + fi + + - name: Resolve Source Repository Name + id: source_repo + run: | + set -euo pipefail + if [[ ! "${SOURCE_REPO}" =~ ^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$ ]]; then + echo "Invalid SOURCE_REPO: ${SOURCE_REPO}. Expected owner/name." >&2 + exit 1 + fi + allowed_match=false + IFS=',' read -ra allowed_repos <<< "${ALLOWED_SOURCE_REPOS}" + for allowed_repo in "${allowed_repos[@]}"; do + allowed_repo="${allowed_repo//[[:space:]]/}" + if [ "${allowed_repo}" = "${SOURCE_REPO}" ]; then + allowed_match=true + break + fi + done + if [ "${allowed_match}" != "true" ]; then + echo "SOURCE_REPO is not allowed for strategy watcher: ${SOURCE_REPO}" >&2 + exit 1 + fi + ref_allowed=false + IFS=',' read -ra allowed_refs <<< "${ALLOWED_SOURCE_REFS}" + for allowed_ref in "${allowed_refs[@]}"; do + allowed_ref="${allowed_ref//[[:space:]]/}" + if [ "${allowed_ref}" = "${SOURCE_REF}" ]; then + ref_allowed=true + break + fi + done + if [ "${ref_allowed}" != "true" ]; then + echo "SOURCE_REF is not allowed for strategy watcher: ${SOURCE_REF}" >&2 + exit 1 + fi + owner="${SOURCE_REPO%%/*}" + repository="${SOURCE_REPO#*/}" + echo "owner=${owner}" >> "$GITHUB_OUTPUT" + echo "repository=${repository}" >> "$GITHUB_OUTPUT" + + - name: Create GitHub App Token For Source Repository + id: source_app_token + if: steps.app_credentials.outputs.available == 'true' + continue-on-error: true + uses: actions/create-github-app-token@v3.2.0 + with: + app-id: ${{ vars.CROSS_REPO_GITHUB_APP_ID }} + private-key: ${{ secrets.CROSS_REPO_GITHUB_APP_PRIVATE_KEY }} + owner: ${{ steps.source_repo.outputs.owner }} + repositories: ${{ steps.source_repo.outputs.repository }} + permission-contents: read + permission-issues: write + + - name: Verify Source Repository Token + env: + SOURCE_APP_TOKEN: ${{ steps.source_app_token.outputs.token }} + run: | + set -euo pipefail + if [ "${SOURCE_REPO}" != "${GITHUB_REPOSITORY}" ] && [ -z "${SOURCE_APP_TOKEN:-}" ]; then + echo "Cross-repository strategy watcher requires CROSS_REPO_GITHUB_APP_ID and CROSS_REPO_GITHUB_APP_PRIVATE_KEY." >&2 + exit 1 + fi + + - name: Checkout Source Metrics + uses: actions/checkout@v6.0.3 + with: + repository: ${{ env.SOURCE_REPO }} + ref: ${{ env.SOURCE_REF }} + path: source + token: ${{ steps.source_app_token.outputs.token || github.token }} + persist-credentials: false + + - name: Run Strategy Optimization Watcher + env: + GH_TOKEN: ${{ steps.source_app_token.outputs.token || github.token }} + STRATEGY_WATCH_SOURCE_ROOT: ${{ github.workspace }}/source + STRATEGY_WATCH_METRICS_PATH: ${{ env.METRICS_PATH }} + STRATEGY_WATCH_SOURCE_REPO: ${{ env.SOURCE_REPO }} + working-directory: bridge + run: | + set -euo pipefail + mkdir -p data/output/strategy_optimization_watcher + python scripts/run_strategy_optimization_watcher.py | tee data/output/strategy_optimization_watcher/result.json + + - name: Upload watcher diagnostics + if: always() + uses: actions/upload-artifact@v7 + with: + name: strategy-optimization-watcher-${{ github.run_id }} + path: bridge/data/output/strategy_optimization_watcher/ + if-no-files-found: warn diff --git a/docs/ai_autonomy_architecture.md b/docs/ai_autonomy_architecture.md index 92265815..41902644 100644 --- a/docs/ai_autonomy_architecture.md +++ b/docs/ai_autonomy_architecture.md @@ -257,6 +257,28 @@ AIAuditBridge 是 QuantStrategyLab 的 AI 审计控制面，负责： 这类优化有价值，但不是无人值守的先决条件。 +#### 3.8 Strategy Optimization Watcher 的 issue-only 安全边界 + +对于 Strategy Optimization Watcher，首批实现只走 issue-only 提案流，不直接触碰策略执行面。 + +推荐流程是： + +1. deterministic trigger 触发 watcher； +2. 生成 evidence bundle； +3. 只创建 optimization issue / task proposal； +4. 经过 authority / registry gate 校验后再决定是否进入下一步； +5. 后续如需执行，再由人工或 CI gate 接管。 + +当前首批实现的安全边界是： + +- 只创建 optimization issue/task； +- 不自动改策略； +- 不调 live 参数； +- 不联网检索； +- 不自动 merge / deploy。 + +这样可以把策略优化先收敛为可审计、可回放的建议流，再逐步扩展到受控执行面。 + --- ## 4. 可落地的阶段性改造计划 diff --git a/scripts/run_strategy_optimization_watcher.py b/scripts/run_strategy_optimization_watcher.py new file mode 100755 index 00000000..bf288eec --- /dev/null +++ b/scripts/run_strategy_optimization_watcher.py @@ -0,0 +1,287 @@ +#!/usr/bin/env python3 +"""Run the issue-only strategy optimization watcher.""" + +from __future__ import annotations + +import copy +import json +import os +from pathlib import Path, PurePosixPath +import re +import subprocess +import sys +from typing import Any, Callable + +ROOT = Path(__file__).resolve().parents[1] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from service.strategy_watch import evaluate_strategy_watch, finding_to_automation_task, issue_for_task, watcher_issue_key # noqa: E402 + +REPO_RE = re.compile(r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$") + + +def parse_bool(value: Any, *, default: bool = False) -> bool: + if value is None: + return default + if isinstance(value, bool): + return value + text = str(value).strip().lower() + if not text: + return default + if text in {"1", "true", "yes", "on"}: + return True + if text in {"0", "false", "no", "off"}: + return False + raise ValueError("boolean value must be one of true/false/yes/no/on/off/1/0") + + +def resolve_input_path( + *, + input_path: str = "", + source_root: str = "", + metrics_path: str = "", +) -> Path | None: + if source_root and metrics_path: + normalized = PurePosixPath(metrics_path.replace("\\", "/")) + if normalized.is_absolute() or ".." in normalized.parts: + raise ValueError("metrics_path must be a relative path inside the source checkout") + root = Path(source_root).resolve() + candidate = (root / Path(*normalized.parts)).resolve() + try: + candidate.relative_to(root) + except ValueError as exc: + raise ValueError("metrics_path resolves outside the source checkout") from exc + return candidate + if not input_path: + return None + candidate = Path(input_path).resolve() + if source_root: + root = Path(source_root).resolve() + try: + candidate.relative_to(root) + except ValueError as exc: + raise ValueError("input path resolves outside the source checkout") from exc + return candidate + + +def load_payload(path: str | Path) -> dict[str, Any]: + payload = json.loads(Path(path).read_text(encoding="utf-8")) + if not isinstance(payload, dict): + raise ValueError("strategy watch input must be a JSON object") + return payload + + +def list_open_issue_urls(repo: str) -> dict[str, str]: + if not REPO_RE.fullmatch(repo): + raise ValueError("repository must be in owner/name form") + page = 1 + open_issues: dict[str, str] = {} + while True: + result = subprocess.run( + ["gh", "api", "--method", "GET", f"/repos/{repo}/issues", "-f", "state=open", "-f", "per_page=100", "-f", f"page={page}"], + check=True, + capture_output=True, + text=True, + ) + try: + issues = json.loads(result.stdout or "[]") + except json.JSONDecodeError as exc: + raise RuntimeError("failed to parse open issue list") from exc + if not isinstance(issues, list) or not issues: + return open_issues + for issue in issues: + if not isinstance(issue, dict) or "pull_request" in issue: + continue + body = str(issue.get("body") or "") + match = re.search(r"", body) + if match and match.group(1) not in open_issues: + open_issues[match.group(1)] = str(issue.get("html_url") or issue.get("url") or "") + if len(issues) < 100: + return open_issues + page += 1 + + +def find_existing_open_issue(repo: str, issue_key: str) -> str: + return list_open_issue_urls(repo).get(issue_key, "") + + +def task_public_summary(task: Any) -> dict[str, Any]: + payload = task.to_dict() + trigger = payload.get("trigger") if isinstance(payload.get("trigger"), dict) else {} + proposed_action = payload.get("proposed_action") if isinstance(payload.get("proposed_action"), dict) else {} + gate_decision = payload.get("gate_decision") if isinstance(payload.get("gate_decision"), dict) else {} + return { + "trigger": { + "source": trigger.get("source", ""), + "kind": trigger.get("kind", ""), + "severity": trigger.get("severity", ""), + "subject": trigger.get("subject", ""), + }, + "proposed_action": { + "action": proposed_action.get("action", ""), + "lane": proposed_action.get("lane", ""), + "target": proposed_action.get("target", ""), + "requires_human_review": proposed_action.get("requires_human_review", True), + }, + "gate_decision": { + "allowed": gate_decision.get("allowed", False), + "human_review_required": gate_decision.get("human_review_required", True), + }, + "status": payload.get("status", ""), + } + + +def comment_github_issue(repo: str, issue_url: str, body: str) -> str: + if not REPO_RE.fullmatch(repo): + raise ValueError("repository must be in owner/name form") + result = subprocess.run( + ["gh", "issue", "comment", issue_url, "--repo", repo, "--body", body], + check=True, + capture_output=True, + text=True, + ) + return result.stdout.strip() + + +def create_github_issue(repo: str, title: str, body: str) -> str: + if not REPO_RE.fullmatch(repo): + raise ValueError("repository must be in owner/name form") + result = subprocess.run( + ["gh", "issue", "create", "--repo", repo, "--title", title, "--body", body], + check=True, + capture_output=True, + text=True, + ) + return result.stdout.strip() + + +def _payload_for_source_repo(payload: dict[str, Any], source_repo: str) -> dict[str, Any]: + if not source_repo: + return payload + normalized = copy.deepcopy(payload) + for key in ("repo", "repository"): + embedded = str(normalized.get(key) or "").strip() + if embedded and embedded != source_repo: + raise ValueError("metrics payload repository does not match validated source repository") + normalized["repo"] = source_repo + raw_snapshots = normalized.get("snapshots") + if isinstance(raw_snapshots, list): + clean_snapshots: list[Any] = [] + for item in raw_snapshots: + if not isinstance(item, dict): + clean_snapshots.append(item) + continue + snapshot = dict(item) + for key in ("repo", "repository"): + embedded = str(snapshot.get(key) or "").strip() + if embedded and embedded != source_repo: + raise ValueError("snapshot repository does not match validated source repository") + snapshot["repo"] = source_repo + clean_snapshots.append(snapshot) + normalized["snapshots"] = clean_snapshots + return normalized + + +def run_watcher( + payload: dict[str, Any], + *, + source_repo: str = "", + dry_run: bool = True, + create_issue: Callable[[str, str, str], str] = create_github_issue, + comment_issue: Callable[[str, str, str], str] = comment_github_issue, + list_issues: Callable[[str], dict[str, str]] = list_open_issue_urls, +) -> dict[str, Any]: + if not dry_run and not source_repo: + raise ValueError("source_repo is required for non-dry-run strategy watcher runs") + if source_repo and not REPO_RE.fullmatch(source_repo): + raise ValueError("source_repo must be in owner/name form") + watch_payload = _payload_for_source_repo(payload, source_repo) + findings = evaluate_strategy_watch(watch_payload) + issues: list[dict[str, Any]] = [] + open_issue_cache: dict[str, dict[str, str]] = {} + for finding in findings: + task = finding_to_automation_task(finding) + issue = issue_for_task(task) + issue_key = watcher_issue_key(task) + repo = source_repo or finding.snapshot.repo + issue_result: dict[str, Any] = { + "repo": repo, + "title": issue["title"], + "task": task_public_summary(task), + "watcher_issue_key": issue_key, + "created": False, + } + if dry_run: + issue_result["dry_run"] = True + else: + try: + if repo not in open_issue_cache: + open_issue_cache[repo] = list_issues(repo) + existing_url = open_issue_cache[repo].get(issue_key, "") + if existing_url: + issue_result["existing_url"] = existing_url + issue_result["comment_url"] = comment_issue(repo, existing_url, issue["body"]) + issue_result["commented"] = True + issue_result["skipped_reason"] = "open issue already exists; appended watcher update" + else: + issue_result["url"] = create_issue(repo, issue["title"], issue["body"]) + open_issue_cache[repo][issue_key] = str(issue_result["url"]) + issue_result["created"] = True + except (ValueError, RuntimeError, subprocess.CalledProcessError) as exc: + issue_result["error"] = str(exc) + issues.append(issue_result) + errors = sum(1 for issue in issues if issue.get("error")) + return { + "status": "partial_error" if errors else "ok", + "dry_run": dry_run, + "findings": len(findings), + "issues": issues, + "errors": errors, + } + + +def main() -> int: + try: + input_path = resolve_input_path( + input_path=os.environ.get("STRATEGY_WATCH_INPUT", "").strip(), + source_root=os.environ.get("STRATEGY_WATCH_SOURCE_ROOT", "").strip(), + metrics_path=os.environ.get("STRATEGY_WATCH_METRICS_PATH", "").strip(), + ) + except ValueError as exc: + print(json.dumps({"status": "error", "error": str(exc)}, sort_keys=True)) + return 2 + if input_path is None: + print(json.dumps({"status": "skipped", "reason": "strategy metrics input not configured"}, sort_keys=True)) + return 0 + if not input_path.exists(): + print(json.dumps({"status": "error", "error": "strategy metrics input not found"}, sort_keys=True)) + return 2 + if not input_path.is_file(): + print(json.dumps({"status": "error", "error": "strategy metrics input is not a file"}, sort_keys=True)) + return 2 + try: + payload = load_payload(input_path) + except (OSError, json.JSONDecodeError, ValueError) as exc: + print(json.dumps({"status": "error", "error": str(exc)}, sort_keys=True)) + return 2 + try: + dry_run = parse_bool(os.environ.get("STRATEGY_WATCH_DRY_RUN"), default=True) + except ValueError as exc: + print(json.dumps({"status": "error", "error": str(exc)}, sort_keys=True)) + return 2 + try: + result = run_watcher( + payload, + source_repo=os.environ.get("STRATEGY_WATCH_SOURCE_REPO", "").strip(), + dry_run=dry_run, + ) + except (ValueError, RuntimeError, subprocess.CalledProcessError) as exc: + print(json.dumps({"status": "error", "error": str(exc)}, sort_keys=True)) + return 2 + print(json.dumps(result, ensure_ascii=False, sort_keys=True)) + return 1 if int(result.get("errors", 0)) > 0 or result.get("status") != "ok" else 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/service/automation_contracts.py b/service/automation_contracts.py new file mode 100644 index 00000000..8fbb2242 --- /dev/null +++ b/service/automation_contracts.py @@ -0,0 +1,127 @@ +"""Minimal shared contracts for automation strategy data.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any + + +def _validate_non_empty(value: str, field_name: str) -> None: + if not str(value).strip(): + raise ValueError(f"{field_name} must be a non-empty string") + + +@dataclass +class TriggerRecord: + source: str + kind: str + severity: str + reason: str + subject: str + metrics: dict[str, Any] = field(default_factory=dict) + evidence: list[Any] = field(default_factory=list) + created_at: float | None = None + + def __post_init__(self) -> None: + _validate_non_empty(self.severity, "severity") + + def to_dict(self) -> dict[str, Any]: + return { + "source": self.source, + "kind": self.kind, + "severity": self.severity, + "reason": self.reason, + "subject": self.subject, + "metrics": dict(self.metrics), + "evidence": list(self.evidence), + "created_at": self.created_at, + } + + +@dataclass +class EvidenceBundle: + summary: str + artifacts: list[Any] = field(default_factory=list) + metrics: dict[str, Any] = field(default_factory=dict) + risks: list[Any] = field(default_factory=list) + + def to_dict(self) -> dict[str, Any]: + return { + "summary": self.summary, + "artifacts": list(self.artifacts), + "metrics": dict(self.metrics), + "risks": list(self.risks), + } + + +@dataclass +class ProposedAction: + action: str + lane: str + target: str + rationale: str + requires_human_review: bool = True + metadata: dict[str, Any] = field(default_factory=dict) + + def __post_init__(self) -> None: + _validate_non_empty(self.action, "action") + + def to_dict(self) -> dict[str, Any]: + return { + "action": self.action, + "lane": self.lane, + "target": self.target, + "rationale": self.rationale, + "requires_human_review": self.requires_human_review, + "metadata": dict(self.metadata), + } + + +@dataclass +class GateDecision: + allowed: bool + reason: str + required_checks: list[Any] = field(default_factory=list) + human_review_required: bool = True + metadata: dict[str, Any] = field(default_factory=dict) + + def to_dict(self) -> dict[str, Any]: + return { + "allowed": self.allowed, + "reason": self.reason, + "required_checks": list(self.required_checks), + "human_review_required": self.human_review_required, + "metadata": dict(self.metadata), + } + + +@dataclass +class AutomationTask: + trigger: TriggerRecord + evidence: EvidenceBundle + proposed_action: ProposedAction + gate_decision: GateDecision + status: str = "proposed" + metadata: dict[str, Any] = field(default_factory=dict) + + def __post_init__(self) -> None: + _validate_non_empty(self.status, "status") + + @property + def is_actionable(self) -> bool: + return ( + self.gate_decision.allowed + and bool(self.proposed_action.action.strip()) + and not self.gate_decision.human_review_required + and not self.proposed_action.requires_human_review + ) + + def to_dict(self) -> dict[str, Any]: + return { + "trigger": self.trigger.to_dict(), + "evidence": self.evidence.to_dict(), + "proposed_action": self.proposed_action.to_dict(), + "gate_decision": self.gate_decision.to_dict(), + "status": self.status, + "metadata": dict(self.metadata), + } diff --git a/service/strategy_optimization_policy.py b/service/strategy_optimization_policy.py new file mode 100644 index 00000000..455c82cd --- /dev/null +++ b/service/strategy_optimization_policy.py @@ -0,0 +1,130 @@ +"""Deterministic policy for strategy optimization watch triggers.""" + +from __future__ import annotations + +from copy import deepcopy +from dataclasses import dataclass, field +from typing import Any + +SEVERITY_NONE = "none" +SEVERITY_MEDIUM = "medium" +SEVERITY_HIGH = "high" + +DEFAULT_METRIC_RULES: dict[str, dict[str, Any]] = { + "sharpe": {"higher_better": True, "relative_drop": 0.05}, + "cagr": {"higher_better": True, "relative_drop": 0.05}, + "calmar": {"higher_better": True, "relative_drop": 0.05}, + "win_rate": {"higher_better": True, "relative_drop": 0.03}, + "max_dd": {"higher_better": False, "absolute_worsening": 0.02}, +} +HIGH_SEVERITY_SIGNAL_COUNT = 2 +HIGH_SEVERITY_MAX_DD_WORSENING = 0.05 + + +@dataclass(frozen=True) +class StrategyOptimizationPolicy: + """Thresholds used by the watcher; deterministic and service-owned.""" + + metric_rules: dict[str, dict[str, Any]] = field(default_factory=lambda: deepcopy(DEFAULT_METRIC_RULES)) + high_severity_signal_count: int = HIGH_SEVERITY_SIGNAL_COUNT + high_severity_max_dd_worsening: float = HIGH_SEVERITY_MAX_DD_WORSENING + + +def _safe_float(value: Any) -> float | None: + if isinstance(value, bool): + return None + try: + return float(value) + except (TypeError, ValueError): + return None + + +def metric_degradation_signals( + current_metrics: dict[str, Any], + baseline_metrics: dict[str, Any], + *, + policy: StrategyOptimizationPolicy | None = None, +) -> list[dict[str, Any]]: + """Return deterministic degradation signals beyond configured thresholds.""" + active_policy = policy or StrategyOptimizationPolicy() + signals: list[dict[str, Any]] = [] + for metric, rule in active_policy.metric_rules.items(): + current = _safe_float(current_metrics.get(metric)) + baseline = _safe_float(baseline_metrics.get(metric)) + if current is None or baseline is None: + continue + delta = current - baseline + signal: dict[str, Any] | None = None + if rule.get("higher_better") is False: + threshold = float(rule.get("absolute_worsening", 0.0)) + if delta > threshold: + signal = { + "metric": metric, + "baseline": baseline, + "current": current, + "delta": delta, + "threshold": threshold, + "reason": f"{metric} worsened by {delta:.4g} > {threshold:.4g}", + } + else: + threshold = float(rule.get("relative_drop", 0.0)) + absolute_threshold = float(rule.get("absolute_drop_when_zero", threshold)) + if baseline == 0: + relative_delta = None + degraded = current < -absolute_threshold + else: + relative_delta = delta / abs(baseline) + degraded = relative_delta < -threshold + if degraded: + reason = ( + f"{metric} dropped below zero by {abs(current):.4g} > {absolute_threshold:.4g}" + if relative_delta is None + else f"{metric} dropped {relative_delta:.1%} beyond {threshold:.1%}" + ) + signal = { + "metric": metric, + "baseline": baseline, + "current": current, + "delta": delta, + "relative_delta": relative_delta, + "threshold": threshold, + "reason": reason, + } + if signal is not None: + signals.append(signal) + return signals + + +def classify_strategy_degradation( + signals: list[dict[str, Any]], + *, + policy: StrategyOptimizationPolicy | None = None, +) -> str: + """Classify watcher severity without involving an LLM.""" + if not signals: + return SEVERITY_NONE + active_policy = policy or StrategyOptimizationPolicy() + if len(signals) >= active_policy.high_severity_signal_count: + return SEVERITY_HIGH + for signal in signals: + if signal.get("metric") == "max_dd" and float(signal.get("delta") or 0.0) >= active_policy.high_severity_max_dd_worsening: + return SEVERITY_HIGH + return SEVERITY_MEDIUM + + +def evaluate_strategy_metrics( + current_metrics: dict[str, Any], + baseline_metrics: dict[str, Any], + *, + policy: StrategyOptimizationPolicy | None = None, +) -> dict[str, Any]: + """Evaluate whether a strategy profile should open an optimization issue.""" + active_policy = policy or StrategyOptimizationPolicy() + signals = metric_degradation_signals(current_metrics, baseline_metrics, policy=active_policy) + severity = classify_strategy_degradation(signals, policy=active_policy) + return { + "should_open_issue": bool(signals), + "severity": severity, + "signals": signals, + "signal_count": len(signals), + } diff --git a/service/strategy_watch.py b/service/strategy_watch.py new file mode 100644 index 00000000..b0162a72 --- /dev/null +++ b/service/strategy_watch.py @@ -0,0 +1,217 @@ +"""Strategy optimization watcher for issue-only automation proposals.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +import hashlib +import json +from typing import Any + +from service.automation_contracts import AutomationTask, EvidenceBundle, GateDecision, ProposedAction, TriggerRecord +from service.strategy_automation_registry import LANE_RESEARCH_BACKLOG, summarize_strategy_registry_context +from service.strategy_optimization_policy import evaluate_strategy_metrics + +WATCHER_SCHEMA_VERSION = "strategy_optimization_watch.v1" +ISSUE_ONLY_ACTION = "open_issue" + + +def _dict_payload(value: Any) -> dict[str, Any]: + return dict(value) if isinstance(value, dict) else {} + + +@dataclass(frozen=True) +class StrategyWatchSnapshot: + repo: str + profile: str + plugin: str = "" + current_metrics: dict[str, Any] = field(default_factory=dict) + baseline_metrics: dict[str, Any] = field(default_factory=dict) + source: str = "" + generated_at: str = "" + + @classmethod + def from_dict(cls, payload: dict[str, Any], *, default_repo: str = "") -> "StrategyWatchSnapshot": + profile = str(payload.get("strategy_profile") or payload.get("profile") or "").strip() + return cls( + repo=str(payload.get("repo") or payload.get("repository") or default_repo).strip(), + profile=profile, + plugin=str(payload.get("plugin") or payload.get("strategy_plugin") or "").strip(), + current_metrics=_dict_payload(payload.get("current_metrics") or payload.get("current")), + baseline_metrics=_dict_payload(payload.get("baseline_metrics") or payload.get("baseline")), + source=str(payload.get("source") or "").strip(), + generated_at=str(payload.get("generated_at") or "").strip(), + ) + + def subject(self) -> str: + parts = [self.repo, self.profile or self.plugin] + return ":".join(part for part in parts if part) + + def to_dict(self) -> dict[str, Any]: + return { + "repo": self.repo, + "profile": self.profile, + "plugin": self.plugin, + "current_metrics": self.current_metrics, + "baseline_metrics": self.baseline_metrics, + "source": self.source, + "generated_at": self.generated_at, + } + + +@dataclass(frozen=True) +class StrategyWatchFinding: + snapshot: StrategyWatchSnapshot + severity: str + signals: list[dict[str, Any]] + registry_context: dict[str, Any] = field(default_factory=dict) + + def to_dict(self) -> dict[str, Any]: + return { + "schema_version": WATCHER_SCHEMA_VERSION, + "snapshot": self.snapshot.to_dict(), + "severity": self.severity, + "signals": self.signals, + "registry_context": self.registry_context, + } + + +def _snapshots_from_payload(payload: dict[str, Any]) -> list[StrategyWatchSnapshot]: + default_repo = str(payload.get("repo") or payload.get("repository") or "").strip() + raw_snapshots = payload.get("snapshots") + if not isinstance(raw_snapshots, list): + raw_snapshots = [payload] + snapshots: list[StrategyWatchSnapshot] = [] + for item in raw_snapshots: + if isinstance(item, dict): + snapshots.append(StrategyWatchSnapshot.from_dict(item, default_repo=default_repo)) + return snapshots + + +def evaluate_strategy_watch(payload: dict[str, Any]) -> list[StrategyWatchFinding]: + """Evaluate metrics payload and return issue-worthy findings only.""" + registry_payload = payload.get("automation_registry") or payload.get("registry") or {} + findings: list[StrategyWatchFinding] = [] + for snapshot in _snapshots_from_payload(payload): + decision = evaluate_strategy_metrics(snapshot.current_metrics, snapshot.baseline_metrics) + if not decision["should_open_issue"]: + continue + context = summarize_strategy_registry_context(registry_payload, snapshot.profile) if snapshot.profile else {} + findings.append( + StrategyWatchFinding( + snapshot=snapshot, + severity=str(decision["severity"]), + signals=list(decision["signals"]), + registry_context=context, + ) + ) + return findings + + +def finding_event_key(finding: StrategyWatchFinding) -> str: + payload = { + "snapshot": finding.snapshot.to_dict(), + "severity": finding.severity, + "signals": finding.signals, + } + raw = json.dumps(payload, ensure_ascii=False, sort_keys=True, default=str).encode("utf-8") + return hashlib.sha256(raw).hexdigest()[:12] + + +def finding_to_automation_task(finding: StrategyWatchFinding) -> AutomationTask: + """Convert a deterministic finding into an issue-only automation task.""" + lane = str(finding.registry_context.get("automation_lane") or LANE_RESEARCH_BACKLOG) + event_key = finding_event_key(finding) + signal_reasons = [str(signal.get("reason") or signal.get("metric") or "metric degraded") for signal in finding.signals] + trigger = TriggerRecord( + source="strategy_optimization_watcher", + kind="strategy_metric_degradation", + severity=finding.severity, + reason="; ".join(signal_reasons) or "strategy metrics degraded", + subject=finding.snapshot.subject(), + metrics=finding.snapshot.current_metrics, + evidence=signal_reasons, + ) + evidence = EvidenceBundle( + summary="Deterministic strategy metrics crossed degradation thresholds.", + artifacts=[finding.snapshot.source] if finding.snapshot.source else [], + metrics={ + "current": finding.snapshot.current_metrics, + "baseline": finding.snapshot.baseline_metrics, + }, + risks=[ + "issue-only: no strategy code, live parameters, broker/order paths, or deployment are changed", + "sandbox backtest evidence is required before any PR can be proposed", + ], + ) + proposed = ProposedAction( + action=ISSUE_ONLY_ACTION, + lane=lane, + target=finding.snapshot.repo, + rationale="Open a research optimization issue for AI diagnosis and sandbox experiment planning.", + requires_human_review=True, + metadata={"profile": finding.snapshot.profile, "plugin": finding.snapshot.plugin, "event_key": event_key}, + ) + gate = GateDecision( + allowed=True, + reason="Issue-only proposal is allowed; code changes and live-impact actions remain gated.", + required_checks=[ + "human review before strategy code changes", + "sandbox backtest before optimization PR", + "strategy registry/authority gate before live impact", + ], + human_review_required=True, + metadata={"issue_only": True, "live_impact_allowed": False}, + ) + return AutomationTask( + trigger=trigger, + evidence=evidence, + proposed_action=proposed, + gate_decision=gate, + metadata={"event_key": event_key}, + ) + + +def watcher_issue_key(task: AutomationTask) -> str: + payload = task.to_dict() + trigger = payload.get("trigger") if isinstance(payload.get("trigger"), dict) else {} + subject = str(trigger.get("subject") or "") + raw = json.dumps({"subject": subject}, ensure_ascii=False, sort_keys=True).encode("utf-8") + return hashlib.sha256(raw).hexdigest()[:16] + + +def issue_for_task(task: AutomationTask) -> dict[str, str]: + """Build a GitHub issue title/body for a strategy optimization task.""" + payload = task.to_dict() + trigger = payload["trigger"] + evidence = payload["evidence"] + action = payload["proposed_action"] + event_key = str(payload.get("metadata", {}).get("event_key") or "") + issue_key = watcher_issue_key(task) + title = f"AI strategy optimization proposal: {trigger.get('subject') or action.get('target') or 'strategy profile'}" + signals = "\n".join(f"- {item}" for item in trigger.get("evidence", [])) or "- Strategy metrics degraded." + checks = "\n".join(f"- [ ] {item}" for item in payload["gate_decision"].get("required_checks", [])) + risks = "\n".join(f"- {item}" for item in evidence.get("risks", [])) + body = "\n".join( + [ + f"", + "## Summary", + str(evidence.get("summary") or "Strategy optimization watcher opened this issue."), + "", + "## Trigger", + f"- Severity: `{trigger.get('severity')}`", + f"- Subject: `{trigger.get('subject')}`", + f"- Event key: `{event_key}`", + "", + "## Signals", + signals, + "", + "## Safety boundary", + risks, + "", + "## Required gates before code/live impact", + checks, + "", + "This watcher only opens an issue. It does not modify strategy code, tune live parameters, merge PRs, or deploy.", + ] + ) + return {"title": title[:240], "body": body} diff --git a/tests/test_automation_contracts.py b/tests/test_automation_contracts.py new file mode 100644 index 00000000..1ecfa1ee --- /dev/null +++ b/tests/test_automation_contracts.py @@ -0,0 +1,184 @@ +"""Tests for service/automation_contracts.py — minimal automation data model.""" + +from __future__ import annotations + +import unittest + +from service.automation_contracts import ( + AutomationTask, + EvidenceBundle, + GateDecision, + ProposedAction, + TriggerRecord, +) + + +class TestAutomationContracts(unittest.TestCase): + def test_to_dict(self) -> None: + trigger = TriggerRecord( + source="watcher", + kind="strategy_update", + severity="high", + reason="threshold exceeded", + subject="strategy-a", + metrics={"score": 0.91}, + evidence=["log-1"], + created_at=123.4, + ) + evidence = EvidenceBundle( + summary="summary", + artifacts=["artifact-1"], + metrics={"latency": 42}, + risks=["regression"], + ) + action = ProposedAction( + action="adjust_threshold", + lane="automation", + target="strategy-a", + rationale="reduce risk", + requires_human_review=False, + metadata={"source": "test"}, + ) + decision = GateDecision( + allowed=True, + reason="checks passed", + required_checks=["unit-tests"], + human_review_required=False, + metadata={"reviewer": "bot"}, + ) + task = AutomationTask( + trigger=trigger, + evidence=evidence, + proposed_action=action, + gate_decision=decision, + status="approved", + metadata={"owner": "team-a"}, + ) + + self.assertEqual( + task.to_dict(), + { + "trigger": { + "source": "watcher", + "kind": "strategy_update", + "severity": "high", + "reason": "threshold exceeded", + "subject": "strategy-a", + "metrics": {"score": 0.91}, + "evidence": ["log-1"], + "created_at": 123.4, + }, + "evidence": { + "summary": "summary", + "artifacts": ["artifact-1"], + "metrics": {"latency": 42}, + "risks": ["regression"], + }, + "proposed_action": { + "action": "adjust_threshold", + "lane": "automation", + "target": "strategy-a", + "rationale": "reduce risk", + "requires_human_review": False, + "metadata": {"source": "test"}, + }, + "gate_decision": { + "allowed": True, + "reason": "checks passed", + "required_checks": ["unit-tests"], + "human_review_required": False, + "metadata": {"reviewer": "bot"}, + }, + "status": "approved", + "metadata": {"owner": "team-a"}, + }, + ) + + def test_mutable_defaults_are_not_shared(self) -> None: + first_trigger = TriggerRecord("a", "k", "high", "r", "s") + second_trigger = TriggerRecord("a", "k", "high", "r", "s") + first_trigger.metrics["x"] = 1 + first_trigger.evidence.append("e1") + self.assertEqual(second_trigger.metrics, {}) + self.assertEqual(second_trigger.evidence, []) + + first_bundle = EvidenceBundle("sum") + second_bundle = EvidenceBundle("sum") + first_bundle.artifacts.append("a1") + first_bundle.metrics["m"] = 2 + first_bundle.risks.append("r1") + self.assertEqual(second_bundle.artifacts, []) + self.assertEqual(second_bundle.metrics, {}) + self.assertEqual(second_bundle.risks, []) + + first_action = ProposedAction("act", "lane", "target", "why") + second_action = ProposedAction("act", "lane", "target", "why") + first_action.metadata["x"] = "y" + self.assertEqual(second_action.metadata, {}) + + first_decision = GateDecision(True, "ok") + second_decision = GateDecision(True, "ok") + first_decision.required_checks.append("c1") + first_decision.metadata["x"] = "y" + self.assertEqual(second_decision.required_checks, []) + self.assertEqual(second_decision.metadata, {}) + + first_task = AutomationTask( + trigger=first_trigger, + evidence=first_bundle, + proposed_action=first_action, + gate_decision=first_decision, + ) + second_task = AutomationTask( + trigger=second_trigger, + evidence=second_bundle, + proposed_action=second_action, + gate_decision=second_decision, + ) + first_task.metadata["x"] = 1 + self.assertEqual(second_task.metadata, {}) + + def test_empty_field_validation(self) -> None: + with self.assertRaises(ValueError): + TriggerRecord("a", "k", "", "r", "s") + with self.assertRaises(ValueError): + ProposedAction("", "lane", "target", "why") + with self.assertRaises(ValueError): + AutomationTask( + trigger=TriggerRecord("a", "k", "high", "r", "s"), + evidence=EvidenceBundle("sum"), + proposed_action=ProposedAction("act", "lane", "target", "why"), + gate_decision=GateDecision(True, "ok"), + status="", + ) + + def test_is_actionable(self) -> None: + trigger = TriggerRecord("a", "k", "high", "r", "s") + evidence = EvidenceBundle("sum") + allowed_task = AutomationTask( + trigger=trigger, + evidence=evidence, + proposed_action=ProposedAction("act", "lane", "target", "why", requires_human_review=False), + gate_decision=GateDecision(True, "ok", human_review_required=False), + ) + blocked_task = AutomationTask( + trigger=trigger, + evidence=evidence, + proposed_action=ProposedAction("act", "lane", "target", "why", requires_human_review=False), + gate_decision=GateDecision(False, "blocked", human_review_required=False), + ) + + human_review_task = AutomationTask( + trigger=trigger, + evidence=evidence, + proposed_action=ProposedAction("act", "lane", "target", "why", requires_human_review=True), + gate_decision=GateDecision(True, "ok", human_review_required=True), + ) + + self.assertTrue(allowed_task.is_actionable) + self.assertFalse(blocked_task.is_actionable) + self.assertFalse(human_review_task.is_actionable) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_run_strategy_optimization_watcher.py b/tests/test_run_strategy_optimization_watcher.py new file mode 100644 index 00000000..88ef9f6b --- /dev/null +++ b/tests/test_run_strategy_optimization_watcher.py @@ -0,0 +1,236 @@ +from __future__ import annotations + +import json +import subprocess +import tempfile +import unittest +from unittest.mock import patch + +from scripts.run_strategy_optimization_watcher import list_open_issue_urls, parse_bool, resolve_input_path, run_watcher + + +class RunStrategyOptimizationWatcherTest(unittest.TestCase): + def test_dry_run_does_not_create_issue(self) -> None: + calls: list[tuple[str, str, str]] = [] + + result = run_watcher( + { + "repo": "QuantStrategyLab/TestStrategies", + "profile": "live", + "current_metrics": {"sharpe": 0.5}, + "baseline_metrics": {"sharpe": 1.0}, + }, + dry_run=True, + create_issue=lambda repo, title, body: calls.append((repo, title, body)) or "https://example.test/1", + comment_issue=lambda repo, url, body: "https://example.test/comment", + list_issues=lambda repo: {}, + ) + + self.assertEqual(result["findings"], 1) + self.assertEqual(result["errors"], 0) + self.assertTrue(result["issues"][0]["dry_run"]) + self.assertNotIn("metrics", result["issues"][0]["task"]["trigger"]) + self.assertEqual(calls, []) + + def test_non_dry_run_uses_source_repo_override(self) -> None: + calls: list[tuple[str, str, str]] = [] + + result = run_watcher( + { + "repo": "QuantStrategyLab/IssueRepo", + "profile": "live", + "current_metrics": {"max_dd": 0.2}, + "baseline_metrics": {"max_dd": 0.1}, + }, + source_repo="QuantStrategyLab/IssueRepo", + dry_run=False, + create_issue=lambda repo, title, body: calls.append((repo, title, body)) or "https://example.test/issue/1", + comment_issue=lambda repo, url, body: "https://example.test/comment", + list_issues=lambda repo: {}, + ) + + self.assertTrue(result["issues"][0]["created"]) + self.assertEqual(result["issues"][0]["url"], "https://example.test/issue/1") + self.assertEqual(calls[0][0], "QuantStrategyLab/IssueRepo") + + def test_non_dry_run_skips_existing_open_issue(self) -> None: + calls: list[tuple[str, str, str]] = [] + payload = { + "repo": "QuantStrategyLab/TestStrategies", + "profile": "live", + "current_metrics": {"sharpe": 0.5}, + "baseline_metrics": {"sharpe": 1.0}, + } + dry_result = run_watcher(payload, dry_run=True) + issue_key = dry_result["issues"][0]["watcher_issue_key"] + + result = run_watcher( + payload, + source_repo="QuantStrategyLab/TestStrategies", + dry_run=False, + create_issue=lambda repo, title, body: calls.append((repo, title, body)) or "https://example.test/new", + comment_issue=lambda repo, url, body: "https://example.test/comment", + list_issues=lambda repo: {issue_key: "https://example.test/existing"}, + ) + + self.assertFalse(result["issues"][0]["created"]) + self.assertEqual(result["issues"][0]["existing_url"], "https://example.test/existing") + self.assertEqual(result["issues"][0]["comment_url"], "https://example.test/comment") + self.assertTrue(result["issues"][0]["commented"]) + self.assertEqual(calls, []) + + def test_resolve_input_path_rejects_metrics_path_traversal(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + with self.assertRaises(ValueError): + resolve_input_path(source_root=tmp, metrics_path="../outside.json") + with self.assertRaises(ValueError): + resolve_input_path(source_root=tmp, metrics_path="/tmp/outside.json") + + def test_resolve_input_path_accepts_source_relative_metrics_path(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + resolved = resolve_input_path(source_root=tmp, metrics_path="data/output/strategy_metrics.json") + + self.assertTrue(str(resolved).endswith("data/output/strategy_metrics.json")) + + def test_find_existing_open_issue_paginates_until_exact_match(self) -> None: + calls: list[list[str]] = [] + first_page = [{"title": f"other-{i}", "body": "", "html_url": f"https://example.test/{i}"} for i in range(100)] + second_page = [{"title": "target", "body": "", "html_url": "https://example.test/target"}] + + def fake_run(cmd: list[str], **kwargs: object) -> subprocess.CompletedProcess[str]: + calls.append(cmd) + page = "2" if "page=2" in cmd else "1" + payload = second_page if page == "2" else first_page + return subprocess.CompletedProcess(cmd, 0, stdout=json.dumps(payload), stderr="") + + with patch("scripts.run_strategy_optimization_watcher.subprocess.run", fake_run): + issues = list_open_issue_urls("QuantStrategyLab/TestStrategies") + + self.assertEqual(issues["abc12345"], "https://example.test/target") + self.assertEqual(len(calls), 2) + self.assertIn("--method", calls[0]) + self.assertIn("GET", calls[0]) + + def test_run_watcher_requires_source_repo_for_non_dry_run(self) -> None: + with self.assertRaises(ValueError): + run_watcher( + { + "repo": "QuantStrategyLab/TestStrategies", + "current_metrics": {"sharpe": 0.5}, + "baseline_metrics": {"sharpe": 1.0}, + }, + dry_run=False, + ) + + def test_run_watcher_rejects_mismatched_source_repo_payload(self) -> None: + with self.assertRaises(ValueError): + run_watcher( + { + "repo": "QuantStrategyLab/Other", + "current_metrics": {"sharpe": 0.5}, + "baseline_metrics": {"sharpe": 1.0}, + }, + source_repo="QuantStrategyLab/TestStrategies", + ) + + def test_run_watcher_uses_validated_source_repo_in_task_summary(self) -> None: + result = run_watcher( + { + "profile": "live", + "current_metrics": {"sharpe": 0.5}, + "baseline_metrics": {"sharpe": 1.0}, + }, + source_repo="QuantStrategyLab/TestStrategies", + dry_run=True, + ) + + self.assertIn("QuantStrategyLab/TestStrategies:live", result["issues"][0]["title"]) + self.assertNotRegex(result["issues"][0]["title"], r"\[[a-f0-9]{12}\]$") + self.assertEqual(result["issues"][0]["task"]["proposed_action"]["target"], "QuantStrategyLab/TestStrategies") + + def test_run_watcher_records_issue_errors_per_finding(self) -> None: + payload = { + "repo": "QuantStrategyLab/TestStrategies", + "snapshots": [ + {"profile": "a", "current_metrics": {"sharpe": 0.5}, "baseline_metrics": {"sharpe": 1.0}}, + {"profile": "b", "current_metrics": {"sharpe": 0.4}, "baseline_metrics": {"sharpe": 1.0}}, + ], + } + + result = run_watcher( + payload, + source_repo="QuantStrategyLab/TestStrategies", + dry_run=False, + create_issue=lambda repo, title, body: (_ for _ in ()).throw(RuntimeError("boom")) if ":a" in title else "https://example.test/new", + comment_issue=lambda repo, url, body: "https://example.test/comment", + list_issues=lambda repo: {}, + ) + + self.assertEqual(result["status"], "partial_error") + self.assertEqual(result["errors"], 1) + self.assertEqual(len(result["issues"]), 2) + + def test_run_watcher_updates_cache_after_create(self) -> None: + create_calls: list[tuple[str, str, str]] = [] + payload = { + "repo": "QuantStrategyLab/TestStrategies", + "snapshots": [ + {"profile": "same", "current_metrics": {"sharpe": 0.5}, "baseline_metrics": {"sharpe": 1.0}}, + {"profile": "same", "current_metrics": {"sharpe": 0.5}, "baseline_metrics": {"sharpe": 1.0}}, + ], + } + + result = run_watcher( + payload, + source_repo="QuantStrategyLab/TestStrategies", + dry_run=False, + create_issue=lambda repo, title, body: create_calls.append((repo, title, body)) or "https://example.test/new", + comment_issue=lambda repo, url, body: "https://example.test/comment", + list_issues=lambda repo: {}, + ) + + self.assertEqual(result["findings"], 2) + self.assertEqual(len(create_calls), 1) + self.assertEqual(result["issues"][1]["existing_url"], "https://example.test/new") + + def test_run_watcher_caches_open_issues_per_repo(self) -> None: + list_calls: list[str] = [] + create_calls: list[tuple[str, str, str]] = [] + payload = { + "repo": "QuantStrategyLab/TestStrategies", + "snapshots": [ + {"profile": "a", "current_metrics": {"sharpe": 0.5}, "baseline_metrics": {"sharpe": 1.0}}, + {"profile": "b", "current_metrics": {"sharpe": 0.4}, "baseline_metrics": {"sharpe": 1.0}}, + ], + } + + result = run_watcher( + payload, + source_repo="QuantStrategyLab/TestStrategies", + dry_run=False, + create_issue=lambda repo, title, body: create_calls.append((repo, title, body)) or "https://example.test/new", + list_issues=lambda repo: list_calls.append(repo) or {}, + ) + + self.assertEqual(result["findings"], 2) + self.assertEqual(list_calls, ["QuantStrategyLab/TestStrategies"]) + self.assertEqual(len(create_calls), 2) + + def test_list_open_issue_urls_fails_closed_on_bad_json(self) -> None: + def fake_run(cmd: list[str], **kwargs: object) -> subprocess.CompletedProcess[str]: + return subprocess.CompletedProcess(cmd, 0, stdout="not-json", stderr="") + + with patch("scripts.run_strategy_optimization_watcher.subprocess.run", fake_run): + with self.assertRaises(RuntimeError): + list_open_issue_urls("QuantStrategyLab/TestStrategies") + + def test_parse_bool_defaults_safely(self) -> None: + self.assertTrue(parse_bool("true")) + self.assertFalse(parse_bool("false")) + self.assertTrue(parse_bool(None, default=True)) + with self.assertRaises(ValueError): + parse_bool("flase") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_strategy_optimization_policy.py b/tests/test_strategy_optimization_policy.py new file mode 100644 index 00000000..c1576f14 --- /dev/null +++ b/tests/test_strategy_optimization_policy.py @@ -0,0 +1,44 @@ +from __future__ import annotations + +import unittest + +from service.strategy_optimization_policy import SEVERITY_HIGH, SEVERITY_MEDIUM, SEVERITY_NONE, evaluate_strategy_metrics + + +class StrategyOptimizationPolicyTest(unittest.TestCase): + def test_detects_multi_metric_degradation_as_high_severity(self) -> None: + result = evaluate_strategy_metrics( + {"sharpe": 0.8, "max_dd": 0.16}, + {"sharpe": 1.0, "max_dd": 0.10}, + ) + + self.assertTrue(result["should_open_issue"]) + self.assertEqual(result["severity"], SEVERITY_HIGH) + self.assertEqual({signal["metric"] for signal in result["signals"]}, {"sharpe", "max_dd"}) + + def test_single_small_degradation_is_medium_severity(self) -> None: + result = evaluate_strategy_metrics({"sharpe": 0.9}, {"sharpe": 1.0}) + + self.assertTrue(result["should_open_issue"]) + self.assertEqual(result["severity"], SEVERITY_MEDIUM) + + def test_baseline_zero_negative_current_is_degradation(self) -> None: + result = evaluate_strategy_metrics({"sharpe": -0.2}, {"sharpe": 0.0}) + + self.assertTrue(result["should_open_issue"]) + self.assertEqual(result["severity"], SEVERITY_MEDIUM) + self.assertEqual(result["signals"][0]["metric"], "sharpe") + + def test_ignores_missing_or_below_threshold_metrics(self) -> None: + result = evaluate_strategy_metrics( + {"sharpe": 0.98, "max_dd": 0.111, "calmar": "not-a-number"}, + {"sharpe": 1.0, "max_dd": 0.10, "calmar": 1.2}, + ) + + self.assertFalse(result["should_open_issue"]) + self.assertEqual(result["severity"], SEVERITY_NONE) + self.assertEqual(result["signals"], []) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_strategy_optimization_watcher_workflow.py b/tests/test_strategy_optimization_watcher_workflow.py new file mode 100644 index 00000000..8d706120 --- /dev/null +++ b/tests/test_strategy_optimization_watcher_workflow.py @@ -0,0 +1,53 @@ +from __future__ import annotations + +from pathlib import Path +import unittest + + +WORKFLOW_PATH = Path(__file__).resolve().parents[1] / ".github/workflows/strategy_optimization_watcher.yml" + + +class StrategyOptimizationWatcherWorkflowTest(unittest.TestCase): + def test_workflow_is_issue_only_and_dry_run_by_default(self) -> None: + text = WORKFLOW_PATH.read_text(encoding="utf-8") + + self.assertIn("name: Strategy Optimization Watcher", text) + self.assertIn("default: true", text) + self.assertIn("STRATEGY_WATCH_DRY_RUN", text) + self.assertIn("scripts/run_strategy_optimization_watcher.py", text) + self.assertIn("permission-issues: write", text) + self.assertNotIn("pull-request", text.lower()) + self.assertNotIn("auto_merge", text.lower()) + self.assertNotIn("deploy", text.lower()) + + def test_workflow_uses_source_metrics_checkout(self) -> None: + text = WORKFLOW_PATH.read_text(encoding="utf-8") + + self.assertIn("SOURCE_REPO", text) + self.assertIn("QuantStrategyLab/CryptoLivePoolPipelines", text) + self.assertIn("STRATEGY_WATCH_ALLOWED_SOURCE_REPOS", text) + self.assertIn("STRATEGY_WATCH_ALLOWED_SOURCE_REFS", text) + self.assertIn("SOURCE_REF is not allowed", text) + self.assertNotIn("vars.STRATEGY_WATCH_SOURCE_REPO || github.repository", text) + self.assertIn("METRICS_PATH", text) + self.assertIn("path: source", text) + self.assertIn("STRATEGY_WATCH_SOURCE_ROOT: ${{ github.workspace }}/source", text) + self.assertIn("STRATEGY_WATCH_METRICS_PATH: ${{ env.METRICS_PATH }}", text) + self.assertIn("actions/create-github-app-token", text) + self.assertIn("format('{0}', inputs.dry_run)", text) + self.assertNotIn("github.event.inputs.dry_run ||", text) + + def test_workflow_fails_closed_for_cross_repo_without_app_token(self) -> None: + text = WORKFLOW_PATH.read_text(encoding="utf-8") + + self.assertIn("^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$", text) + self.assertIn("Verify Source Repository Token", text) + self.assertIn("${SOURCE_REPO}" + '" != "' + "${GITHUB_REPOSITORY}", text) + self.assertIn("Cross-repository strategy watcher requires", text) + self.assertIn("SOURCE_REPO is not allowed", text) + self.assertIn("owner=${owner}", text) + self.assertIn("owner: ${{ steps.source_repo.outputs.owner }}", text) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_strategy_watch.py b/tests/test_strategy_watch.py new file mode 100644 index 00000000..5a4b68b3 --- /dev/null +++ b/tests/test_strategy_watch.py @@ -0,0 +1,81 @@ +from __future__ import annotations + +import unittest + +from service.strategy_watch import evaluate_strategy_watch, finding_to_automation_task, issue_for_task + + +class StrategyWatchTest(unittest.TestCase): + def test_degraded_snapshot_becomes_issue_only_task(self) -> None: + findings = evaluate_strategy_watch( + { + "repo": "QuantStrategyLab/TestStrategies", + "snapshots": [ + { + "strategy_profile": "mean_reversion_live", + "plugin": "mean_reversion", + "current_metrics": {"sharpe": 0.7, "max_dd": 0.18}, + "baseline_metrics": {"sharpe": 1.0, "max_dd": 0.1}, + "source": "data/output/strategy_metrics.json", + } + ], + } + ) + + self.assertEqual(len(findings), 1) + task = finding_to_automation_task(findings[0]) + payload = task.to_dict() + self.assertFalse(task.is_actionable) + self.assertEqual(payload["proposed_action"]["action"], "open_issue") + self.assertTrue(payload["proposed_action"]["requires_human_review"]) + self.assertTrue(payload["gate_decision"]["human_review_required"]) + self.assertFalse(payload["gate_decision"]["metadata"]["live_impact_allowed"]) + + def test_malformed_metrics_snapshot_is_ignored_without_crashing(self) -> None: + findings = evaluate_strategy_watch( + { + "repo": "QuantStrategyLab/TestStrategies", + "snapshots": [ + {"profile": "bad", "current_metrics": "oops", "baseline_metrics": {"sharpe": 1.0}}, + {"profile": "live", "current_metrics": {"sharpe": 0.5}, "baseline_metrics": {"sharpe": 1.0}}, + ], + } + ) + + self.assertEqual(len(findings), 1) + self.assertEqual(findings[0].snapshot.profile, "live") + + def test_healthy_snapshot_creates_no_finding(self) -> None: + findings = evaluate_strategy_watch( + { + "repo": "QuantStrategyLab/TestStrategies", + "current_metrics": {"sharpe": 1.01, "max_dd": 0.10}, + "baseline_metrics": {"sharpe": 1.0, "max_dd": 0.1}, + } + ) + + self.assertEqual(findings, []) + + def test_issue_body_states_safety_boundary(self) -> None: + finding = evaluate_strategy_watch( + { + "repo": "QuantStrategyLab/TestStrategies", + "profile": "live", + "current_metrics": {"sharpe": 0.8}, + "baseline_metrics": {"sharpe": 1.0}, + } + )[0] + + issue = issue_for_task(finding_to_automation_task(finding)) + + self.assertIn("AI strategy optimization proposal", issue["title"]) + self.assertNotRegex(issue["title"], r"\[[a-f0-9]{12}\]$") + self.assertIn("Event key", issue["body"]) + self.assertIn("