From 6eed6989f952352d0f1cec79a390a85523f8f2ca Mon Sep 17 00:00:00 2001 From: abgyjaguo <213890245+abgyjaguo@users.noreply.github.com> Date: Tue, 11 Aug 2026 15:29:08 +0800 Subject: [PATCH 1/2] feat(contract): expose versioned skill interface --- .cursor/rules/backtest.mdc | 7 ++ .gitignore | 2 + CLAUDE.md | 3 + README.en.md | 4 + README.md | 6 + SKILL.md | 64 ++++++++++ agents/hermes-loader.md | 3 + agents/openclaw-loader.md | 3 + scripts/backtest_dag.py | 146 ++++++++++++++++++++++ tests/fixtures/factor-panel.json | 1 + tests/fixtures/market-bar-ohlcv-only.json | 1 + tests/fixtures/market-bar-trading.json | 1 + tests/test_backtest_dag.py | 29 +++++ 13 files changed, 270 insertions(+) create mode 100644 .cursor/rules/backtest.mdc create mode 100644 .gitignore create mode 100644 CLAUDE.md create mode 100644 agents/hermes-loader.md create mode 100644 agents/openclaw-loader.md create mode 100644 scripts/backtest_dag.py create mode 100644 tests/fixtures/factor-panel.json create mode 100644 tests/fixtures/market-bar-ohlcv-only.json create mode 100644 tests/fixtures/market-bar-trading.json create mode 100644 tests/test_backtest_dag.py diff --git a/.cursor/rules/backtest.mdc b/.cursor/rules/backtest.mdc new file mode 100644 index 0000000..23cf6d4 --- /dev/null +++ b/.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. diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..d2f4ae0 --- /dev/null +++ b/.gitignore @@ -0,0 +1,2 @@ +__pycache__/ +tests/tmp-output*/ diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 0000000..e3dfc64 --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1,3 @@ +# Backtest skill adapter + +Follow the canonical [SKILL.md](SKILL.md), including the explicit envelope and CLI contract. diff --git a/README.en.md b/README.en.md index a37c03e..3d502a0 100644 --- a/README.en.md +++ b/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. diff --git a/README.md b/README.md index c3081f8..92690ff 100644 --- a/README.md +++ b/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 项健康度自检。 diff --git a/SKILL.md b/SKILL.md index 1efc3db..73612e5 100644 --- a/SKILL.md +++ b/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: [] +--- + +--- +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]` 模拟交易出来，得到净值 / 回撤 / 换手 / 分组收益。从"统计相关"走向"可执行收益"的关键一步。 ## 核心规则 diff --git a/agents/hermes-loader.md b/agents/hermes-loader.md new file mode 100644 index 0000000..2f632fd --- /dev/null +++ b/agents/hermes-loader.md @@ -0,0 +1,3 @@ +# Hermes adapter + +Install the root [SKILL.md](../SKILL.md) as the native skill entrypoint. diff --git a/agents/openclaw-loader.md b/agents/openclaw-loader.md new file mode 100644 index 0000000..132ec8c --- /dev/null +++ b/agents/openclaw-loader.md @@ -0,0 +1,3 @@ +# OpenClaw adapter + +Install the root [SKILL.md](../SKILL.md) as the native skill entrypoint. diff --git a/scripts/backtest_dag.py b/scripts/backtest_dag.py new file mode 100644 index 0000000..f3cf0ad --- /dev/null +++ b/scripts/backtest_dag.py @@ -0,0 +1,146 @@ +#!/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})$") +SHA = "sha256:" + "0" * 64 +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 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(data).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) -> 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} + sleeves, previous, series, turnover = [], defaultdict(float), [], [] + for i, date in enumerate(dates): + if len(sleeves) >= horizon: sleeves.pop(0) + if i + 1 < len(dates) and date in signals: + ranked = sorted(signals[date].items(), key=lambda x: (-x[1], x[0])); count = max(1, math.ceil(len(ranked) * top_pct)); nxt = dates[i + 1] + eligible = [symbol for symbol, _ in ranked[:count] if symbol in universe[nxt] and bars[symbol, nxt]["trade_status"] != 1 and bars[symbol, nxt]["close"] < bars[symbol, nxt]["limit_up"] * .99] + if eligible: sleeves.append({symbol: 1 / horizon / len(eligible) for symbol in eligible}) + position = defaultdict(float) + for sleeve in sleeves: + for symbol, weight in sleeve.items(): position[symbol] += weight + gross = 0.0 + if i: + prior = dates[i - 1] + for symbol, weight in previous.items(): + if (symbol, date) in bars and (symbol, prior) in bars: gross += weight * (bars[symbol, date]["close"] / bars[symbol, prior]["close"] - 1) + delta = sum(abs(position[s] - previous[s]) for s in set(position) | set(previous)) / 2 + net = gross - delta * 2 * fee_bps / 1e4 if i else 0.0 + series.append({"timestamp": date, "return": net, "turnover": delta, "nav": (series[-1]["nav"] if series else 1.0) * (1 + net)}) + turnover.append(delta); previous = position + 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} + 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": SHA}] + lineage = {"sources": [{"profile": "factor-panel", "version": "1.0.0", "artifact_ref": "artifact://input/factor-panel", "sha256": SHA}, {"profile": "market-bar", "version": "1.0.0", "artifact_ref": "artifact://input/market-bar", "sha256": 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": sum(x["turnover"] for x in series) * 2 * fee_bps / 1e4, "assumptions": {"factor_id": factor_id, "horizon": horizon, "top_pct": top_pct, "fee_bps": fee_bps, "execution": "T+1-open-signal/close-to-close-return"}, "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) + 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() diff --git a/tests/fixtures/factor-panel.json b/tests/fixtures/factor-panel.json new file mode 100644 index 0000000..897e3f1 --- /dev/null +++ b/tests/fixtures/factor-panel.json @@ -0,0 +1 @@ +{"$contract":{"profile":"factor-panel","profile_version":"1.0.0"},"meta":{"dataset_id":"fixture-factor","producer":"fixture","generated_at":"2026-08-10T09:30:00Z","timezone":"UTC"},"schema":{"fields":{"instrument_id":{},"timestamp":{},"factor_id":{},"value":{},"direction":{},"frequency":{},"universe":{},"missing_policy":{},"neutralization":{}}},"payload":{"records":[{"instrument_id":"AAA","timestamp":"2026-08-01T09:30:00Z","factor_id":"fixture-alpha","value":2,"direction":"higher-is-better","frequency":"1d","universe":"fixture","missing_policy":"exclude","neutralization":"none"},{"instrument_id":"BBB","timestamp":"2026-08-01T09:30:00Z","factor_id":"fixture-alpha","value":1,"direction":"higher-is-better","frequency":"1d","universe":"fixture","missing_policy":"exclude","neutralization":"none"}]}} diff --git a/tests/fixtures/market-bar-ohlcv-only.json b/tests/fixtures/market-bar-ohlcv-only.json new file mode 100644 index 0000000..1070caa --- /dev/null +++ b/tests/fixtures/market-bar-ohlcv-only.json @@ -0,0 +1 @@ +{"$contract":{"profile":"market-bar","profile_version":"1.0.0"},"meta":{"dataset_id":"bad","producer":"fixture","generated_at":"2026-08-10T09:30:00Z","currency":"CNY"},"schema":{"fields":{"instrument_id":{},"timestamp":{},"open":{},"high":{},"low":{},"close":{},"volume":{},"frequency":{},"adjustment":{},"calendar":{}}},"payload":{"native":{"raw_records":[]},"records":[{"instrument_id":"AAA","timestamp":"2026-08-01T09:30:00Z","open":1,"high":1,"low":1,"close":1,"volume":1,"frequency":"1d","adjustment":"none","calendar":"SYNX"}]}} diff --git a/tests/fixtures/market-bar-trading.json b/tests/fixtures/market-bar-trading.json new file mode 100644 index 0000000..1b3b40f --- /dev/null +++ b/tests/fixtures/market-bar-trading.json @@ -0,0 +1 @@ +{"$contract":{"profile":"market-bar","profile_version":"1.0.0"},"meta":{"dataset_id":"fixture-market","producer":"fixture","generated_at":"2026-08-10T09:30:00Z","timezone":"UTC","currency":"CNY","calendar":"SYNX","frequency":"1d"},"schema":{"fields":{"instrument_id":{},"timestamp":{},"open":{},"high":{},"low":{},"close":{},"volume":{},"frequency":{},"adjustment":{},"calendar":{},"trade_status":{},"limit_up":{},"limit_down":{},"limit_basis":{}}},"payload":{"native":{"raw_records":[{"instrument_id":"AAA","timestamp":"2026-08-01T09:30:00Z","open":10,"high":11,"low":9,"close":10,"volume":100,"frequency":"1d","adjustment":"none","calendar":"SYNX","trade_status":0,"limit_up":11,"limit_down":9,"limit_basis":"close"},{"instrument_id":"BBB","timestamp":"2026-08-01T09:30:00Z","open":10,"high":11,"low":9,"close":10,"volume":100,"frequency":"1d","adjustment":"none","calendar":"SYNX","trade_status":0,"limit_up":11,"limit_down":9,"limit_basis":"close"},{"instrument_id":"AAA","timestamp":"2026-08-02T09:30:00Z","open":10,"high":12,"low":9,"close":11,"volume":100,"frequency":"1d","adjustment":"none","calendar":"SYNX","trade_status":0,"limit_up":12,"limit_down":8,"limit_basis":"close"},{"instrument_id":"BBB","timestamp":"2026-08-02T09:30:00Z","open":10,"high":11,"low":8,"close":9,"volume":100,"frequency":"1d","adjustment":"none","calendar":"SYNX","trade_status":0,"limit_up":12,"limit_down":8,"limit_basis":"close"}]},"records":[{"instrument_id":"AAA","timestamp":"2026-08-01T09:30:00Z","open":10,"high":11,"low":9,"close":10,"volume":100,"frequency":"1d","adjustment":"none","calendar":"SYNX","trade_status":0,"limit_up":11,"limit_down":9,"limit_basis":"close"},{"instrument_id":"BBB","timestamp":"2026-08-01T09:30:00Z","open":10,"high":11,"low":9,"close":10,"volume":100,"frequency":"1d","adjustment":"none","calendar":"SYNX","trade_status":0,"limit_up":11,"limit_down":9,"limit_basis":"close"},{"instrument_id":"AAA","timestamp":"2026-08-02T09:30:00Z","open":10,"high":12,"low":9,"close":11,"volume":100,"frequency":"1d","adjustment":"none","calendar":"SYNX","trade_status":0,"limit_up":12,"limit_down":8,"limit_basis":"close"},{"instrument_id":"BBB","timestamp":"2026-08-02T09:30:00Z","open":10,"high":11,"low":8,"close":9,"volume":100,"frequency":"1d","adjustment":"none","calendar":"SYNX","trade_status":0,"limit_up":12,"limit_down":8,"limit_basis":"close"}]}} diff --git a/tests/test_backtest_dag.py b/tests/test_backtest_dag.py new file mode 100644 index 0000000..df5c8b2 --- /dev/null +++ b/tests/test_backtest_dag.py @@ -0,0 +1,29 @@ +import copy +import importlib.util +import json +import tempfile +import unittest +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +spec = importlib.util.spec_from_file_location("dag", ROOT / "scripts" / "backtest_dag.py") +dag = importlib.util.module_from_spec(spec); spec.loader.exec_module(dag) + +class DagTest(unittest.TestCase): + def setUp(self): + f = ROOT / "tests" / "fixtures"; self.factor = json.loads((f / "factor-panel.json").read_text()); self.market = json.loads((f / "market-bar-trading.json").read_text()) + def test_valid_outputs_are_deterministic(self): + with tempfile.TemporaryDirectory() as temp: + out = Path(temp); backtest, evaluation = dag.run(self.factor, self.market, out, "fixture", 1, .5, 15, "fixture-alpha") + self.assertEqual(backtest["$contract"]["profile"], "backtest-result"); self.assertEqual(evaluation["$contract"]["profile"], "evaluation-result") + self.assertTrue((out / "return-series.json").is_file()); self.assertNotEqual(evaluation["payload"]["records"][0]["lineage"]["sources"][0]["profile"], "evaluation-result") + def test_ohlcv_only_fixture_fails(self): + bad = json.loads((ROOT / "tests" / "fixtures" / "market-bar-ohlcv-only.json").read_text()) + with self.assertRaises(dag.ContractError): dag.validate_market(bad) + def test_multiple_factor_ids_and_native_mismatch_fail(self): + factor = copy.deepcopy(self.factor); factor["payload"]["records"][1]["factor_id"] = "other" + with self.assertRaises(dag.ContractError): dag.validate_factor(factor, None) + market = copy.deepcopy(self.market); market["payload"]["native"]["raw_records"][0]["close"] = 99 + with self.assertRaises(dag.ContractError): dag.validate_market(market) + +if __name__ == "__main__": unittest.main() From 14c07397d8ebb18dffdcc8620a27f669301bbd6c Mon Sep 17 00:00:00 2001 From: abgyjaguo <213890245+abgyjaguo@users.noreply.github.com> Date: Tue, 11 Aug 2026 15:45:12 +0800 Subject: [PATCH 2/2] fix(backtest): price execution and artifact lineage --- scripts/backtest_dag.py | 61 ++++++++++++++++++++++---------------- tests/test_backtest_dag.py | 27 +++++++++++++++++ 2 files changed, 63 insertions(+), 25 deletions(-) diff --git a/scripts/backtest_dag.py b/scripts/backtest_dag.py index f3cf0ad..537020e 100644 --- a/scripts/backtest_dag.py +++ b/scripts/backtest_dag.py @@ -11,7 +11,6 @@ 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})$") -SHA = "sha256:" + "0" * 64 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"} @@ -27,6 +26,14 @@ def load(path: str) -> dict: 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) @@ -89,10 +96,10 @@ def index(items, source): 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(data).hexdigest() + 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) -> tuple[dict, dict]: +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) @@ -100,34 +107,38 @@ def run(factor_doc: dict, market_doc: dict, out: Path, strategy_id: str, horizon 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} - sleeves, previous, series, turnover = [], defaultdict(float), [], [] + # 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): - if len(sleeves) >= horizon: sleeves.pop(0) - if i + 1 < len(dates) and date in signals: - ranked = sorted(signals[date].items(), key=lambda x: (-x[1], x[0])); count = max(1, math.ceil(len(ranked) * top_pct)); nxt = dates[i + 1] - eligible = [symbol for symbol, _ in ranked[:count] if symbol in universe[nxt] and bars[symbol, nxt]["trade_status"] != 1 and bars[symbol, nxt]["close"] < bars[symbol, nxt]["limit_up"] * .99] - if eligible: sleeves.append({symbol: 1 / horizon / len(eligible) for symbol in eligible}) - position = defaultdict(float) - for sleeve in sleeves: - for symbol, weight in sleeve.items(): position[symbol] += weight - gross = 0.0 - if i: - prior = dates[i - 1] - for symbol, weight in previous.items(): - if (symbol, date) in bars and (symbol, prior) in bars: gross += weight * (bars[symbol, date]["close"] / bars[symbol, prior]["close"] - 1) - delta = sum(abs(position[s] - previous[s]) for s in set(position) | set(previous)) / 2 - net = gross - delta * 2 * fee_bps / 1e4 if i else 0.0 - series.append({"timestamp": date, "return": net, "turnover": delta, "nav": (series[-1]["nav"] if series else 1.0) * (1 + net)}) - turnover.append(delta); previous = position + 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} - 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": SHA}] - lineage = {"sources": [{"profile": "factor-panel", "version": "1.0.0", "artifact_ref": "artifact://input/factor-panel", "sha256": SHA}, {"profile": "market-bar", "version": "1.0.0", "artifact_ref": "artifact://input/market-bar", "sha256": SHA}], "evidence_refs": ["evidence://input/factor-panel", "evidence://input/market-bar"]} + 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": sum(x["turnover"] for x in series) * 2 * fee_bps / 1e4, "assumptions": {"factor_id": factor_id, "horizon": horizon, "top_pct": top_pct, "fee_bps": fee_bps, "execution": "T+1-open-signal/close-to-close-return"}, "bias_controls": ["T+1", "limit-and-suspension-filters", "explicit-inputs"], "lineage": lineage}]}, "quality": {"status": "pass", "checks": ["strict-input-contract", "deterministic-engine"], "warnings": []}} + 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": []}} @@ -137,7 +148,7 @@ def run(factor_doc: dict, market_doc: dict, out: Path, strategy_id: str, horizon 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) + 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") diff --git a/tests/test_backtest_dag.py b/tests/test_backtest_dag.py index df5c8b2..6101c67 100644 --- a/tests/test_backtest_dag.py +++ b/tests/test_backtest_dag.py @@ -1,6 +1,9 @@ import copy +import hashlib import importlib.util import json +import subprocess +import sys import tempfile import unittest from pathlib import Path @@ -17,6 +20,30 @@ def test_valid_outputs_are_deterministic(self): out = Path(temp); backtest, evaluation = dag.run(self.factor, self.market, out, "fixture", 1, .5, 15, "fixture-alpha") self.assertEqual(backtest["$contract"]["profile"], "backtest-result"); self.assertEqual(evaluation["$contract"]["profile"], "evaluation-result") self.assertTrue((out / "return-series.json").is_file()); self.assertNotEqual(evaluation["payload"]["records"][0]["lineage"]["sources"][0]["profile"], "evaluation-result") + self.assertEqual(evaluation["payload"]["records"][0]["lineage"]["sources"][0]["sha256"], "sha256:" + hashlib.sha256((out / "backtest-result.json").read_bytes()).hexdigest()) + self.assertEqual(backtest["payload"]["records"][0]["assumptions"]["return_series_sha256"], "sha256:" + hashlib.sha256((out / "return-series.json").read_bytes()).hexdigest()) + def test_open_execution_and_untradable_exit_delay(self): + market = copy.deepcopy(self.market) + for row in market["payload"]["records"] + market["payload"]["native"]["raw_records"]: + if row["instrument_id"] == "AAA" and row["timestamp"].startswith("2026-08-02"): + row.update(open=20, close=21, limit_up=30, limit_down=1) + extra = {"instrument_id":"AAA","timestamp":"2026-08-03T09:30:00Z","open":30,"high":31,"low":29,"close":30,"volume":100,"frequency":"1d","adjustment":"none","calendar":"SYNX","trade_status":1,"limit_up":40,"limit_down":20,"limit_basis":"close"} + market["payload"]["records"].append(copy.deepcopy(extra)); market["payload"]["native"]["raw_records"].append(copy.deepcopy(extra)) + extra["timestamp"] = "2026-08-04T09:30:00Z"; extra.update(open=40, high=41, low=39, close=40, trade_status=0, limit_down=30) + market["payload"]["records"].append(copy.deepcopy(extra)); market["payload"]["native"]["raw_records"].append(copy.deepcopy(extra)) + with tempfile.TemporaryDirectory() as temp: + _, evaluation = dag.run(self.factor, market, Path(temp), "fixture", 1, .5, 0, "fixture-alpha") + events = json.loads((Path(temp) / "return-series.json").read_text()) + self.assertEqual(events[1]["entry_price"], 20) + self.assertEqual(events[2]["exit_count"], 0) + self.assertEqual(events[3]["exit_price"], 40) + def test_cli_lineage_hashes_input_file_bytes(self): + with tempfile.TemporaryDirectory() as temp: + out = Path(temp) / "output"; fixture = ROOT / "tests" / "fixtures" + subprocess.run([sys.executable, str(ROOT / "scripts" / "backtest_dag.py"), "--factor", str(fixture / "factor-panel.json"), "--market", str(fixture / "market-bar-trading.json"), "--output-dir", str(out), "--strategy-id", "fixture", "--horizon", "1", "--top-pct", "0.5", "--fee-bps", "15", "--factor-id", "fixture-alpha"], check=True, capture_output=True, text=True) + result = json.loads((out / "backtest-result.json").read_text()) + hashes = [x["sha256"] for x in result["payload"]["records"][0]["lineage"]["sources"]] + self.assertEqual(hashes, ["sha256:" + hashlib.sha256((fixture / "factor-panel.json").read_bytes()).hexdigest(), "sha256:" + hashlib.sha256((fixture / "market-bar-trading.json").read_bytes()).hexdigest()]) def test_ohlcv_only_fixture_fails(self): bad = json.loads((ROOT / "tests" / "fixtures" / "market-bar-ohlcv-only.json").read_text()) with self.assertRaises(dag.ContractError): dag.validate_market(bad)