fix: 发布真实 T+1 回测执行与版本化结果接口

.cursor/rules/backtest.mdc

@@ -0,0 +1,7 @@
+---
+description: Use the canonical skill-backtest DAG contract.
+globs: ["**/backtest_dag.py"]
+alwaysApply: false
+---
+
+Read [SKILL.md](SKILL.md). Invoke only the explicit DAG CLI; do not invent defaults or relax its contract checks.

.gitignore

@@ -0,0 +1,2 @@
+__pycache__/
+tests/tmp-output*/

CLAUDE.md

@@ -0,0 +1,3 @@
+# Backtest skill adapter
+
+Follow the canonical [SKILL.md](SKILL.md), including the explicit envelope and CLI contract.

README.en.md

@@ -1,5 +1,9 @@
 # skill-backtest
 
+## Executable DAG interface
+
+`scripts/backtest_dag.py` accepts `factor-panel@1.0.0` and `market-bar@1.0.0` JSON envelopes and requires explicit strategy, horizon, top percentile, fee, and output arguments. It rejects multiple factors, neutral direction, missing tradability or limit evidence, and non-bijective native/canonical records; it emits `backtest-result@1.0.0`, `evaluation-result@1.0.0`, and a hashed internal return-series artifact. Research and education only; not investment advice.
+
 [简体中文](./README.md) | [English](./README.en.md)
 
 Not a backtest framework, but a **standard protocol for cross-section long-only backtesting**: T+1 open execution, Top equal weight, 15bp two-way fee, limit-up/down exclusion, 4-panel diagnostic chart, 5-item health check.

README.md

@@ -1,5 +1,11 @@
 # skill-backtest
 
+Parameters: horizon, top percentile, and fee basis points are explicit. This is for research and education only and does not constitute investment advice.
+
+## 可执行 DAG 接口
+
+`scripts/backtest_dag.py` 接收 `factor-panel@1.0.0` 与 `market-bar@1.0.0` JSON envelope，要求显式传入策略、持有期、Top 比例、费率和输出目录。它拒绝多因子、neutral 方向、缺少交易状态/涨跌停依据及原生记录与 canonical 记录非严格一一对应的输入；输出 `backtest-result@1.0.0`、`evaluation-result@1.0.0` 与带哈希的内部收益序列 artifact。仅供研究和教育，不构成投资建议。
+
 [简体中文](./README.md) | [English](./README.en.md)
 
 不是回测框架，而是**截面多头回测的标准协议**：T+1 开盘成交、Top 等权、双边 15bp、涨跌停剔除、四联诊断图、5 项健康度自检。

SKILL.md

@@ -1,5 +1,53 @@
 ---
 name: backtest
+description: Run a deterministic cross-sectional long-only backtest DAG from versioned factor-panel and market-bar envelopes, with strict T+1 execution, tradability evidence, return-series artifacts, and evaluation outputs. Use when an agent must execute or validate this backtest contract.
+license: GPL-3.0-only
+supported-runtimes: [cursor, claude-code, codex, hermes, openclaw]
+author: abgyjaguo
+metadata:
+  organization: quantskills
+  organization_url: https://github.com/quantskills
+  repository: skill-backtest
+  repository_url: https://github.com/quantskills/skill-backtest
+  project_type: skill
+  collection: backtesting-trading
+  creator: abgyjaguo
+  maintainer: abgyjaguo
+quantSkills:
+  schema_version: 2.1.0
+  organization: quantskills
+  organization_url: https://github.com/quantskills
+  repository: skill-backtest
+  repository_url: https://github.com/quantskills/skill-backtest
+  project_type: skill
+  license: GPL-3.0-only
+  maintainer: abgyjaguo
+  collection: backtesting-trading
+  catalog: {category: "05", subcategory: 05.backtest-engine}
+  workflow:
+    primary_stage: backtesting
+    workflow_stages: [data-ingestion, backtesting, evaluation]
+  tags: [backtest, cross-section, long-only, deterministic]
+  summary_zh: 提供带严格输入证据校验和 T+1 执行约束的横截面多头回测 DAG。
+  summary_en: Runs deterministic cross-sectional long-only backtests with strict evidence validation and T+1 execution.
+  status: active
+  validation_level: verified
+  maintainer_type: community
+  platforms: [cursor, claude-code, codex, hermes, openclaw]
+  interface:
+    mode: structured
+    envelope: {name: quantskills-envelope, version: 1.0.0}
+    inputs:
+      - {profile: factor-panel, version_range: ">=1.0.0 <2.0.0", required: true}
+      - {profile: market-bar, version_range: ">=1.0.0 <2.0.0", required: true}
+    outputs:
+      - {profile: backtest-result, version: 1.0.0}
+      - {profile: evaluation-result, version: 1.0.0}
+    adapters: []
+---
+<!-- Legacy declaration retained below as historical text; canonical metadata is above. -->
+---
+name: backtest
 description: Use when an agent needs a standard cross-sectional long-only backtest
   protocol with T+1 open execution, top-bucket equal weighting, fees, limit-up or
   limit-down exclusions, benchmark comparison, NAV curves, drawdown, IC, and diagnostic
@@ -85,6 +133,22 @@ quantSkills:
 
 # Backtest
 
+This skill is for research and education only and does not constitute investment advice. Parameters are explicit: horizon, top percentile, and fee basis points.
+
+## Deterministic DAG entrypoint
+
+Run `scripts/backtest_dag.py` only with explicit factor, market, strategy, horizon,
+top-percentile, fee, and output arguments. It accepts `factor-panel@1.0.0` and
+`market-bar@1.0.0`, rejects ambiguous factor IDs and market data without preserved
+trading-status and limit evidence, and writes `backtest-result@1.0.0`,
+`evaluation-result@1.0.0`, and a hashed internal return-series artifact. It never
+selects a default factor, fixture, date, or current time.
+
+```bash
+python scripts/backtest_dag.py --factor factor.json --market market.json --output-dir result \
+  --strategy-id strategy-id --horizon 5 --top-pct 0.10 --fee-bps 15 --factor-id factor-id
+```
+
 > 把信号 `[date × symbol]` 模拟交易出来，得到净值 / 回撤 / 换手 / 分组收益。从"统计相关"走向"可执行收益"的关键一步。
 
 ## 核心规则

agents/hermes-loader.md

@@ -0,0 +1,3 @@
+# Hermes adapter
+
+Install the root [SKILL.md](../SKILL.md) as the native skill entrypoint.

agents/openclaw-loader.md

@@ -0,0 +1,3 @@
+# OpenClaw adapter
+
+Install the root [SKILL.md](../SKILL.md) as the native skill entrypoint.

scripts/backtest_dag.py

@@ -0,0 +1,157 @@
+#!/usr/bin/env python3
+"""Deterministic, fail-closed backtest DAG for the backtest skill."""
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import math
+import re
+from collections import defaultdict
+from pathlib import Path
+
+RFC3339 = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:Z|[+-]\d{2}:\d{2})$")
+FACTOR_FIELDS = {"instrument_id", "timestamp", "factor_id", "value", "direction", "frequency", "universe", "missing_policy", "neutralization"}
+MARKET_FIELDS = {"instrument_id", "timestamp", "open", "high", "low", "close", "volume", "frequency", "adjustment", "calendar", "trade_status", "limit_up", "limit_down", "limit_basis"}
+
+
+class ContractError(ValueError): pass
+
+
+def load(path: str) -> dict:
+    try:
+        value = json.loads(Path(path).read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError) as exc: raise ContractError(f"cannot read {path}") from exc
+    if not isinstance(value, dict): raise ContractError("envelope must be an object")
+    return value
+
+
+def file_sha256(path: str | Path) -> str:
+    return "sha256:" + hashlib.sha256(Path(path).read_bytes()).hexdigest()
+
+
+def document_sha256(document: dict) -> str:
+    return "sha256:" + hashlib.sha256(json.dumps(document, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode()).hexdigest()
+
+
+def require(value, message):
+    if not value: raise ContractError(message)
+
+
+def field_names(doc: dict) -> set[str]:
+    fields = doc.get("schema", {}).get("fields")
+    require(isinstance(fields, dict), "schema.fields must be an object")
+    return set(fields)
+
+
+def records(doc: dict, profile: str) -> list[dict]:
+    c = doc.get("$contract", {})
+    require(c.get("profile") == profile and c.get("profile_version") == "1.0.0", f"expected {profile}@1.0.0")
+    rows = doc.get("payload", {}).get("records")
+    require(isinstance(rows, list) and rows, f"{profile} records missing")
+    require(all(isinstance(x, dict) for x in rows), f"{profile} records must be objects")
+    return rows
+
+
+def timestamp(value: object) -> bool: return isinstance(value, str) and RFC3339.fullmatch(value) is not None
+
+
+def validate_factor(doc: dict, selected_id: str | None) -> tuple[list[dict], str, int]:
+    rows = records(doc, "factor-panel")
+    require(FACTOR_FIELDS <= field_names(doc), "factor-panel schema fields incomplete")
+    require(all(FACTOR_FIELDS <= set(r) and timestamp(r["timestamp"]) for r in rows), "invalid factor record")
+    factor_ids = {r["factor_id"] for r in rows}
+    require(len(factor_ids) == 1, "factor-panel must contain one globally unique factor_id")
+    factor_id = next(iter(factor_ids)); require(not selected_id or selected_id == factor_id, "--factor-id does not match factor-panel")
+    directions = {r["direction"] for r in rows}; require(len(directions) == 1, "factor direction must be consistent")
+    direction = directions.pop(); require(direction in {"higher-is-better", "lower-is-better"}, "neutral factor direction is unsupported")
+    seen = set()
+    for row in rows:
+        key = (row["instrument_id"], row["timestamp"]); require(key not in seen, "duplicate factor key"); seen.add(key)
+        require(isinstance(row["value"], (int, float)) and not isinstance(row["value"], bool) and math.isfinite(row["value"]), "factor value must be finite")
+    return rows, factor_id, -1 if direction == "lower-is-better" else 1
+
+
+def validate_market(doc: dict) -> list[dict]:
+    rows = records(doc, "market-bar")
+    require(MARKET_FIELDS <= field_names(doc), "market-bar must declare trade_status, limit_up, limit_down, and limit_basis")
+    native = doc.get("payload", {}).get("native", {}); raw = native.get("raw_records") if isinstance(native, dict) else None
+    require(isinstance(raw, list) and raw, "market-bar payload.native.raw_records is required")
+    def index(items, source):
+        result = {}
+        for row in items:
+            require(isinstance(row, dict) and MARKET_FIELDS <= set(row) and timestamp(row["timestamp"]), f"invalid {source} market record")
+            key = (row["instrument_id"], row["timestamp"]); require(key not in result, f"duplicate {source} market key"); result[key] = row
+        return result
+    canonical, original = index(rows, "canonical"), index(raw, "native")
+    require(canonical.keys() == original.keys(), "native and canonical market keys must be a strict 1:1 mapping")
+    for key, row in canonical.items():
+        other = original[key]
+        require(all(row[k] == other[k] for k in ("open", "high", "low", "close", "volume", "frequency", "calendar", "adjustment", "limit_up", "limit_down", "limit_basis", "trade_status")), "native market record differs from canonical record")
+        require(all(isinstance(row[k], (int, float)) and not isinstance(row[k], bool) for k in ("open", "high", "low", "close", "volume", "limit_up", "limit_down")), "market prices must be numeric")
+        require(row["frequency"] == doc.get("meta", {}).get("frequency", row["frequency"]) and row["calendar"] == doc.get("meta", {}).get("calendar", row["calendar"]), "market metadata inconsistent")
+    return rows
+
+
+def artifact(path: Path, value: object) -> tuple[str, str]:
+    data = json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode()
+    path.write_bytes(data + b"\n")
+    return f"artifact://{path.name}", "sha256:" + hashlib.sha256(path.read_bytes()).hexdigest()
+
+
+def run(factor_doc: dict, market_doc: dict, out: Path, strategy_id: str, horizon: int, top_pct: float, fee_bps: float, selected_id: str | None = None, factor_sha: str | None = None, market_sha: str | None = None) -> tuple[dict, dict]:
+    require(horizon > 0 and 0 < top_pct <= 1 and fee_bps >= 0, "invalid explicit parameters")
+    factors, factor_id, sign = validate_factor(factor_doc, selected_id); market = validate_market(market_doc)
+    bars = {(x["instrument_id"], x["timestamp"]): x for x in market}; signals = defaultdict(dict)
+    for x in factors:
+        require((x["instrument_id"], x["timestamp"]) in bars, "factor key has no market-bar")
+        signals[x["timestamp"]][x["instrument_id"]] = sign * x["value"]
+    dates = sorted({x["timestamp"] for x in market}); universe = {d: {x["instrument_id"] for x in market if x["timestamp"] == d} for d in dates}
+    # A sleeve is purchased at the next session's open, held until its scheduled
+    # open exit, and remains held if a suspension or limit-down prevents selling.
+    cash, sleeves, series, turnover, fees, previous_nav = 1.0, [], [], [], 0.0, 1.0
+    for i, date in enumerate(dates):
+        opened, closed, entry_prices, exit_prices, trade_value = 0, 0, [], [], 0.0
+        for sleeve in list(sleeves):
+            bar = bars.get((sleeve["symbol"], date))
+            if i >= sleeve["exit_index"] and bar and bar["trade_status"] != 1 and bar["close"] > bar["limit_down"] * 1.01:
+                value = sleeve["shares"] * bar["open"]; cost = value * fee_bps / 1e4; cash += value - cost; fees += cost; trade_value += value; exit_prices.append(bar["open"]); sleeves.remove(sleeve); closed += 1
+        if i and dates[i - 1] in signals:
+            ranked = sorted(signals[dates[i - 1]].items(), key=lambda x: (-x[1], x[0])); count = max(1, math.ceil(len(ranked) * top_pct))
+            eligible = [symbol for symbol, _ in ranked[:count] if symbol in universe[date] and bars[symbol, date]["trade_status"] != 1 and bars[symbol, date]["close"] < bars[symbol, date]["limit_up"] * .99]
+            allocation = min(cash, previous_nav / horizon)
+            if eligible and allocation > 0:
+                each = allocation / len(eligible)
+                for symbol in eligible:
+                    bar = bars[symbol, date]; cost = each * fee_bps / 1e4; shares = (each - cost) / bar["open"]
+                    cash -= each; fees += cost; trade_value += each; entry_prices.append(bar["open"]); sleeves.append({"symbol": symbol, "shares": shares, "exit_index": i + horizon}); opened += 1
+        nav = cash + sum(s["shares"] * bars[s["symbol"], date]["close"] for s in sleeves if (s["symbol"], date) in bars)
+        day_turnover = trade_value / (2 * previous_nav) if previous_nav else 0.0
+        series.append({"timestamp": date, "return": nav / previous_nav - 1, "turnover": day_turnover, "nav": nav, "entry_price": entry_prices[0] if entry_prices else None, "exit_price": exit_prices[0] if exit_prices else None, "entry_count": opened, "exit_count": closed})
+        turnover.append(day_turnover); previous_nav = nav
+    ref, digest = artifact(out / "return-series.json", series)
+    navs = [x["nav"] for x in series]; returns = [x["return"] for x in series]
+    annual = navs[-1] ** (252 / len(navs)) - 1; mean = sum(returns) / len(returns); variance = sum((x - mean) ** 2 for x in returns) / max(1, len(returns) - 1)
+    metrics = {"annual_return": annual, "sharpe": mean / math.sqrt(variance) * math.sqrt(252) if variance else 0.0, "max_drawdown": min(n / max(navs[:j + 1]) - 1 for j, n in enumerate(navs)), "annual_turnover": sum(turnover) / len(turnover) * 252}
+    factor_sha, market_sha = factor_sha or document_sha256(factor_doc), market_sha or document_sha256(market_doc)
+    generated = factor_doc["meta"]["generated_at"]; provenance = [{"provider": factor_doc["meta"]["producer"], "dataset": factor_doc["meta"]["dataset_id"], "raw_ref": "artifact://input/factor-panel", "raw_sha256": factor_sha}, {"provider": market_doc["meta"]["producer"], "dataset": market_doc["meta"]["dataset_id"], "raw_ref": "artifact://input/market-bar", "raw_sha256": market_sha}]
+    lineage = {"sources": [{"profile": "factor-panel", "version": "1.0.0", "artifact_ref": "artifact://input/factor-panel", "sha256": factor_sha}, {"profile": "market-bar", "version": "1.0.0", "artifact_ref": "artifact://input/market-bar", "sha256": market_sha}], "evidence_refs": ["evidence://input/factor-panel", "evidence://input/market-bar"]}
+    common = {"generated_at": generated, "as_of": generated, "timezone": factor_doc["meta"].get("timezone", "UTC"), "currency": market_doc["meta"]["currency"], "calendar": market_doc["meta"].get("calendar", market[0]["calendar"]), "provenance": provenance}
+    backtest_fields = {k: {"type": t, "nullable": False} for k, t in {"strategy_id":"string","period_start":"string","period_end":"string","return_series_ref":"string","costs":"number","assumptions":"object","bias_controls":"array","lineage":"object"}.items()}; backtest_fields["period_start"]["format"] = backtest_fields["period_end"]["format"] = "date"; backtest_fields["costs"]["unit"] = "currency"
+    backtest = {"$contract": {"envelope": "quantskills-envelope", "envelope_version": "1.0.0", "profile": "backtest-result", "profile_version": "1.0.0"}, "meta": {"dataset_id": strategy_id, "producer": "skill-backtest", **common}, "schema": {"primary_key": ["strategy_id", "period_start", "period_end"], "fields": backtest_fields}, "payload": {"native": {"provider": "skill-backtest", "raw_records": series}, "records": [{"strategy_id": strategy_id, "period_start": dates[0][:10], "period_end": dates[-1][:10], "return_series_ref": ref, "costs": fees, "assumptions": {"factor_id": factor_id, "horizon": horizon, "top_pct": top_pct, "fee_bps": fee_bps, "entry_price": "open", "exit_price": "open", "holding_mark": "close", "return_series_sha256": digest}, "bias_controls": ["T+1", "limit-and-suspension-filters", "explicit-inputs"], "lineage": lineage}]}, "quality": {"status": "pass", "checks": ["strict-input-contract", "deterministic-engine"], "warnings": []}}
+    backtest_ref, backtest_digest = artifact(out / "backtest-result.json", backtest)
+    evaluation_fields = {k: {"type": t, "nullable": False} for k, t in {"subject_id":"string","evaluated_at":"string","sample_start":"string","sample_end":"string","metrics":"object","verdict":"string","evidence":"array","lineage":"object"}.items()}; evaluation_fields["evaluated_at"]["format"] = "date-time"; evaluation_fields["sample_start"]["format"] = evaluation_fields["sample_end"]["format"] = "date"
+    evaluation = {"$contract": {"envelope": "quantskills-envelope", "envelope_version": "1.0.0", "profile": "evaluation-result", "profile_version": "1.0.0"}, "meta": {"dataset_id": strategy_id + "-evaluation", "producer": "skill-backtest", **common}, "schema": {"primary_key": ["subject_id", "evaluated_at"], "fields": evaluation_fields}, "payload": {"native": {"provider": "skill-backtest", "raw_records": [metrics]}, "records": [{"subject_id": strategy_id, "evaluated_at": generated, "sample_start": dates[0][:10], "sample_end": dates[-1][:10], "metrics": metrics, "verdict": "pass", "evidence": [ref], "lineage": {"sources": [{"profile": "backtest-result", "version": "1.0.0", "artifact_ref": backtest_ref, "sha256": backtest_digest}], "evidence_refs": [ref]}}]}, "quality": {"status": "pass", "checks": ["derived-from-backtest-result"], "warnings": []}}
+    return backtest, evaluation
+
+
+def main():
+    p = argparse.ArgumentParser(); p.add_argument("--factor", required=True); p.add_argument("--market", required=True); p.add_argument("--output-dir", required=True); p.add_argument("--strategy-id", required=True); p.add_argument("--horizon", required=True, type=int); p.add_argument("--top-pct", required=True, type=float); p.add_argument("--fee-bps", required=True, type=float); p.add_argument("--factor-id")
+    a = p.parse_args(); out = Path(a.output_dir); require(not out.exists(), "output directory must not already exist"); out.mkdir(parents=True)
+    try: backtest, evaluation = run(load(a.factor), load(a.market), out, a.strategy_id, a.horizon, a.top_pct, a.fee_bps, a.factor_id, file_sha256(a.factor), file_sha256(a.market))
+    except Exception:
+        out.rmdir(); raise
+    (out / "evaluation-result.json").write_text(json.dumps(evaluation, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
+    print(json.dumps({"backtest_result": str(out / "backtest-result.json"), "evaluation_result": str(out / "evaluation-result.json")}, sort_keys=True))
+
+if __name__ == "__main__": main()