如何利用LangChain深度代理构建多智能体交易研究系统【完整手册】

与单纯的动量策略相比，这种基于体积数据的策略在最大回撤幅度方面确实有所改善，但这种改进所带来的代价并不十分吸引人：开发版本的夏普比率下降了0.0976，年化复合增长率降低了大约两个百分点，而年度交易次数则增加了4.38次。

在代理人开始提出修改建议之前，这些信息是非常有用的。初始策略并不是作为一个只需要稍加改进就能使用的完美方案交给他们的；这个策略本身就已经存在明显的缺陷，他们必须面对这些问题。

同样的基准测试数据也被用于验证过程，并与开发结果一起被保存到了BENCHMARKS.md文件中。这样一来，后续的代理人就可以将自己的修改方案与固定的参考策略进行对比，而不仅仅是根据当前最受欢迎的版本来评判自己的工作成果。

4. 在独立的子进程中运行每种策略

虽然共享引擎统一了性能计算的方式，但生成的策略代码仍然需要在某个地方被执行。

如果直接在主研究流程中执行这些代码，它们就会访问那里已经加载的所有资源，包括API凭证以及我们特意隔离在研究流程之外的数据集。因此，每个实验都将在一个独立的临时进程中运行，这个进程只会包含进行具体评估所需的文件。

首先，我们需要创建在这个独立进程中运行的脚本：

RUNNER = '''
"""用于运行策略的独立进程。拥有自己的进程空间、临时沙盒环境以及干净无杂的数据环境。」"""
import sys, json, importlib.util, traceback

def main():
    strat, params_json, data_dir, split, cost_bps = sys.argv[1:6]
    import engine
    d = engine.load_split(data_dir, split)
    spec = importlib.util.spec_from_file_location("strategy", strat)
    mod = importlib.util.module_from_spec(spec); spec.loader.exec_module(mod)
    w = mod.target_weights(d, **json.loads(params_json))
    bt = engine.backtest(w, d["returns"], cost_bps=float(cost_bps))
    ev = bt.loc[d["eval_start"]:]
    bench = d["returns"]["SPY"].loc[d["eval_start"]:] if "SPY" in d["returns"] else None
    print(json.dumps({"ok": True, "metrics": engine.metrics(ev, bench),
                      "equity": [round(float(x), 6) for x in (1+ev["ret']).cumprod().tolist ],
                      "dates": [str(x.date()) for x in ev.index]}))

if __name__ == "__main__":
    try: main()
    except Exception: print(json.dumps({"ok": False, "error": traceback.format_exc(limit=3)}))
'''
(ROOT/"runner.py").write_text(RUNNER)

def isolated_environment(sandbox):

    required = ["PATH","SYSTEMROOT","WINDIR","COMSPEC","PATHEXT","VIRTUAL_ENV","CONDA_PREFIX","CONDA_DEFAULT_ENV","LD_LIBRARY_PATH",
                "DYLD_library_PATH","LANG","LC_ALL"]

    env = {name: os.environ[name] for name in required if name in os.environ}

    env.update({
        "HOME": str(sandbox),
        "USERPROFILE": str(sandbox),
        "TEMP": str(sandbox),
        "TMP": str(sandbox),
        "TMPDIR": str(sandbox),
        "PYTHONHASHSEED": "1",
        "PYTHONUTF8": "1",
    })

    return env

def run_isolated(strategy_path, params, split, cost_bps=10.0, timeout=600):
    sandbox = Path(tempfile.mkdtemp(prefix="strat_"))
    (sandbox/"data").mkdir()
    for f in ["adj_close","close","volume"]:
        shutil.copy(WS/"data"/f"{split}_{f}.parquet", sandbox/"data")
    shutil.copy(WS/"data"/"splits.json", sandbox/"data")
    shutil.copyROOT/"engine.py", sandbox); shutil.copy ROOT/"runner.py", sandbox)
    shutil.copy(strategy_path, sandbox/"strategy.py")
    try:
        p = subprocess.run([sys.executable, "runner.py", "strategy.py", json.dumps(params),
                            "data", split, str(cost_bps)],
                           capture_output=True, text=True, cwd=sandbox, timeout=timeout,
                           env=isolated_environment(sandbox))
        if not p.stdout.strip():
            return {"ok": False, "error": (p.stderr or "no output")[-400:]}
        return json.loads(p.stdout)
    except subprocess.TimeoutExpired:
        return {"ok": False, "error": f"在{timeout}s后超时"}
    finally:
        shutil.rmtree(sandbox, ignore_errors=True)

在每次运行过程中，run_isolated()会创建一个临时目录，并仅将所需的研究或验证文件以及engine.py、runner.py和正在被评估的策略文件放入该目录中。此外，它还会为子进程构建一个规模较小的环境，而不是直接复制父进程的环境。

因此，生成的策略虽然能够获取计算投资组合权重所需的数据，但它并不需要访问EODHD、OpenAI、LangSmith或保留数据。

这种设计初衷是实现研究流程之间的隔离，并非为了提供操作系统级别的安全防护。生成的代码仍然是在当前用户账户下运行的普通Python进程。这样做的目的是确保策略执行过程中不会意外访问到敏感信息或未公开的研究数据，而不是为了防止恶意代码的攻击。

5. 验证执行结果的准确性及数据访问边界

在正式使用这条执行路径之前，有兩点需要进行检查。

首先，在隔离环境中运行的策略所产生的结果必须与直接使用engine.py执行的相同逻辑所得到的结果完全一致。否则，我们就相当于使用了两种不同的评估标准。

我们将使用“成交量-动量”基准测试来验证这种一致性。

其次，我们会故意运行一个检测程序，该程序会查找那些可能包含敏感信息的环境变量以及保留文件或私有文件。

(WS/"strategies"/"parity_check.py").write_text('''import pandas as pd
def target_weights(data, mom_window=126, top_n=3, vol_short=20, vol_long=120, vol_ratio_min=1.0):
    adj, cls, vol = data["adj_close"], data["close"], data["volume"]
    mom = adj.pct_change(mom_window); dv = cls*vol
    ratio = dv.rolling(vol_short).mean()/dv.rolling.vol_long).mean()
    ok = (mom>0)&(ratio>vol_ratio_min)
    dates = pd.DatetimeIndex(adj.index.to_series().resample("ME").last().dropna())
    w = pd.DataFrame(0.0, index=dates, columns=adj.columns)
    for d in dates:
        picks = mom.loc[d][ok.loc[d]].dropna().nlargest(top_n).index
        if len(picks): w.loc[d,picks]=1.0/len(picks)
    return w
''')
iso = run_isolated(WS/"strategies"/"parity_check.py", {"mom_window":126,"top_n":3}, "dev")
d = DATA["dev"]
inp = metrics(backtest(volume_momentum(d, 126, 3), d["returns")).loc[d["eval_start"]:],
              d["returns"]["SPY"].loc[d["eval_start"]:])
assert iso["metrics"]["sharpe"] == inp["sharpe"], "隔离环境与主线程得到的结果不一致"
print("验证通过:", iso["metrics"]["sharpe"])

PROBE = f'''import os, glob
def target_weights(data, **k):
    keys = [x for x in os.environ if any(t in x for t in ("KEY","TOKEN","SECRET"))]
    files = glob.glob(r"{PRIVATE}/*") + glob.glob(r"{WS}/data/holdout_*")
    raise RuntimeError(f"可访问的敏感文件有：{len(files)}")
'''
(WS/"strategies"/"probe.py").write_text(PROBE)
msg = run_isolated(WS/"strategies"/"probe.py", {}, "dev")["error"].strip().split("\n")[-1]
print("检测结果:", msg)
assert "KEYS=[]" in msg, "沙箱环境中无法访问任何敏感信息"
assert "REACHABLE_SENSITIVE FILES=0" in msg, "沙箱环境中无法访问保留文件或私有文件"

检测结果均通过：

奇偶性检查合格：0.4387
探测结果显示：运行错误：KEYS为空，可访问的敏感文件数量为0

无论是通过隔离路径还是直接路径进行测试，所得到的开发阶段夏普比率均为0.4387，因此这两种方法得出的结果是一致的。探测还发现，在子环境中不存在任何凭证变量，也没有任何分阶段的私有文件或保留文件。

创建实验与决策层

现在，回测引擎为每种策略提供了相同的评估流程。但我们仍然需要控制重复实验过程中发生的各种情况。

如果某个代理可以无限次地测试新的配置方案、忽略失败的试验结果，或者在之前的版本尚未被审查完毕之前就直接切换到新的策略版本，那么研究过程就很可能会偏向于产生最佳结果的那个方向。因此，下一层系统将跟踪每一个实验过程，严格执行固定的研究预算，并要求每个新版本的策略都必须经过相同的评估流程后才能开始后续的测试。

1. 创建实验注册表

我们将首先创建一个记录系统中测试过的所有配置方案的注册表。

REGISTRY = WS / "registry.csv"
DECISIONS = WS / "decisions.jsonl"
MAX_CONFIGS = 12
COLS = ["version","run","status","params","note","dev_cagr","dev_sharpe","dev_sortino",
        "dev_max_dd","dev_turnover","val_cagr","val_sharpe","val_max/dd","dev_cagr_20bps","error"]

def _used(version):
    if not REGISTRY.exists(): return 0
    return int((pd.read_csv(REGISTRY)["version"] == version).sum())

def _decisions():
    if not DECISIONSexists(): return []
    return [json.loads(l) for l in DECISIONS.read_text().splitlines() if l.strip()]

def _stage_ok(version):
    """在v(N-1)版本通过评估并得到最终决定之前，vN版本不能开始测试。”“
    if not (version.startswith("v") and version[1:].isdigit()): return True, ""
    n = int-Version[1:])
    if n <= 1: return True, ""
    prev = f"v{n-1}"
    if not REGISTRY.exists() or _used(prev) == 0:
        return False, f"阶段关卡：{prev}版本没有测试记录。请先完成对{prev}版本的测试。”
    reg = pd.read_csv(REGISTRY)
    if reg[(reg.version == prev) & (reg.status == "ok")].empty:
        return False, f"阶段关卡：{prev}版本没有成功的测试结果。”
    if not (WS/"reviews"/f"{prev}.md").exists():
        return False, f"阶段关卡：/reviews/{prev}.md文件不存在。请先获取评审意见。”
    if not any(d["version"] == prev for d in _decisions()):
        return False, f"阶段关卡：{prev}版本尚未有任何决策记录。请先执行record_decision操作。”
    return True, ""

MAX_CONFIGS = 12这一限制意味着在任何策略版本中，能够被测试的配置方案数量是固定的。这一点非常重要，因为如果验证数据被过度使用，也可能会影响实验结果的准确性。如果代理可以无限制地尝试不同的参数组合，并始终选择在验证过程中表现最好的那个组合，那么验证数据本身就会逐渐成为另一个优化的目标。

阶段门的控制机制涉及另一个不同的问题。新版本的测试无法简单地开始，因为代理可能持有其他不同的意见。在测试v2之前，v1必须至少完成一次成功的运行、经过评审人员的评估，并且有记录在案的决策结果。对于v3来说，也需要遵循同样的流程。

因此，版本的更新流程如下：

这样的流程使得研究步骤能够通过代码来强制执行，而不再需要依赖协调者来记住这些步骤。

2. 创建研究工具

代理们将通过三个LangChain工具与这一层进行交互。

其中最重要的是sweep()这个工具。它是代理获取官方回测结果的唯一途径。

from langchain.tools import tool

@tool
def sweep(version: str, grid_json: str, note: str = "") -> str:
    """在一次调用中，对多个参数设置组合进行策略的回测。

    version   : 文件名后缀，例如"v1"表示strategies/v1.py文件
    grid_json : 参数对象的JSON列表，例如[{"top_n":3},{"top_n":4}]
    note      : 对此次回测操作的简要说明

    会在独立的子进程中运行每个配置。返回按验证夏普比率排序的CSV表格。每个版本最多允许进行12次配置测试，这些数据会累积起来。所有结果都会被写入registry.csv文件中，包括失败的结果。在v(N-1)的回测、评审和决策完成之前，vN是无法继续进行的。
    """
    ok, why = _stage_ok(version)
    if not ok: return f"error: {why}"
    used = _used.version)
    try:
        grid = json.loads(grid_json)
        if isinstance(grid, dict): grid = [grid]
    except Exception as e:
        return f"error: grid_json不是有效的JSON数据 ({e}")
    if used + len(grid) > MAX_CONFIGS:
        return f"error: 预算不足。在{version}上已经使用了{used}/{MAX_CONFIGS}个配置，而你还要求再使用{len(grid)}个配置。"
    path = WS/"strategies"/f"{version}.py"
    if not path.exists():
        return f"error: {path.name}文件不存在，请先创建它。"

    rows = []
    for i, params in enumerate(grid, start=used + 1):
        row = {"version": version, "run": i, "note": note,
               "params": json.dumps(params, separators="", ":"))}
        dev = run_isolated(path, params, "dev")
        if not dev["ok"]:
            row.update(status="error", error=dev["error"].strip().split("\n")[-1][:150])
            rows.append(row); continue
        val = run_isolated(path, params, "val")
        c20 = run_isolated(path, params, "dev", cost_bps=20.0)
        dm, vm = dev["metrics"], val["metrics"]
        row.update(status="ok", dev_cagr=dm["cagr"], dev_sharpe=dm["sharpe"],
                   dev_sortino=dm["sortino"], dev_max_dd=dm["max/dd"],
                   dev_turnover=dm["ann_turnover"], val_cagr=vm["cagr"],
                   val_sharpe=vm["sharpe"], val_max_dd=vm["max-dd"],
                   dev_cagr_20bps=c20["metrics"]["cagr"] if c20["ok"] else None)
        tag = f"{version}_run{i}"
        (WS/"results"/f"{tag}.json").write_text(json.dumps({"params": params, "dev": dm, "val": vm}, indent=2))
        eq = pd.Series(dev["equity"], index=pd.to_datetime/dev["dates"]))
        plt.figure(figsize=(8,3)); plt.plot(eq); plt.yscale("log"); plt.title(tag)
        plt.tight_layout(); plt.savefig(WS/"results"/f"{tag}.png", dpi=90); plt.close("all")
        rows.append(row)

    df = pd.DataFrame(rows).reindex(columns=COLS)
    df.to_csv(REGISTRY, mode="a", header=not REGISTRY.exists(), index=False)
    out = df.dropcolumns=["version","note")).round(3).dropna(axis=1, how="all")
    if "val_sharpe" in out:
        out = out.sort_values("val_sharpe", ascending=False, na_position="last")
    return out.to_csv(index=False)

@tool
def readRegistry(version: str = "") -> str:
    """目前所有测试结果都以CSV格式保存在其中，包括被接受的结果和被拒绝的结果。可以通过传递版本号来过滤结果。"""
    if not REGISTRY.exists(): return "empty"
    r = pd.read_csv(REGistry)
    if version: r = r[r["version"] == version]
    return r[["version","run","status","params","dev_sharpe","dev_sortino",
              "dev_max_dd","val_sharpe","val_max-dd","error"]].to_csv(index=False)

@tool
def record_decision(version: str, champion: str, rationale: str, params_json: str) -> str:
    """记录某个版本的最终决策结果。在开始测试下一个版本之前，必须先完成这个步骤。

    version    : 刚刚被评审的版本号，例如"v2"
    champion  ：根据选择规则确定最终的胜出版本
    rationale ：说明选择该版本的依据以及具体的决定因素
    params_json: 胜出版本的参数设置，以JSON格式提供
    """
    if any(d["version"] == version for d in _decisions()):
        return f"error: 已经存在关于{version}的决策结果，无法覆盖它。"
    rec = {"version": version, "champion": champion, "rationale": rationale,
           "params": json.loads(params_json), "ts": time.time()}
    with DECISIONS.open("a") as f:
        f.write(json.dumps(rec) + "\n")
    return f"决策结果已记录。当前的胜出版本是{champion}"

对于每种配置，sweep()会通过我们刚刚构建的隔离评估路径来执行开发测试和验证流程。同时，它还会在交易成本增加20个基点的条件下重新运行开发测试，这样评估工具就能判断某个结果是否对默认的10个基点假设特别敏感。

成功的运行会生成相应的指标数据、JSON格式的结果文件以及股票价格曲线图。而失败的运行虽然不会从研究记录中消失，但仍然会被保存到registry.csv文件中。这意味着策略工程师不能悄悄地修改一些有问题的配置，然后只展示最终成功的那个版本。

另外两个工具的设计则相对简单。readRegistry()允许代理程序查看被记录下来的数据，而record_decision()则用于生成每个版本的正式评估结果。一旦某个决策被记录下来，就无法通过再次调用同一工具来覆盖它。

3. 修改策略选择规则

注册系统虽然记录了发生了什么，但我们仍然需要明确什么样的改进才算有效。

如果我们等到看到实验结果之后再决定哪些指标才是重要的，那么这些选择标准本身就可能会成为优化过程的一部分。因此，在任何代理程序生成的版本被运行之前，我们就会先修改这个选择规则。

SELECTION_RULE = """
# 版本选择规则（在任何版本被运行之前就已经确定）

只有当一个挑战者满足以下三个条件时，才能取代当前的冠军版本：
1. 其验证阶段的夏普比率不低于当前冠军版本的夏普比率；
2. 其验证阶段的最大回撤幅度在当前冠军版本的2个百分点范围内；
3. 其开发阶段的年收益增长率不超过当前冠军版本的20%。

如果多个挑战者都满足这些条件，那么当前的冠军版本将继续保留。较新的版本不会自动取代旧版本；较高的开发阶段夏普比率并不是判断标准之一。
"""
(WS/"SELECTION_RULE.md").write_text(SELECTION RULE)

def select_champion(challenger, incumbent, name_c, name_i):
    if incumbent is None: return name_c, "没有当前的冠军版本"
    checks = [("validation Sharpe not worse",
               challenger["val_sharpe"] >= incumbent["val_sharpe"],
              ("validation drawdown within 2pp",
               challenger["val_max_dd"] >= incumbent["val_max-dd"] - 0.02),
              ("turnover within +20%",
               challenger["dev_turnover"] <= incumbent["dev_turnover"] * 1.20)]
    failed = [n for n, ok in checks if not ok]
    if failed:
        return name_i, "当前的冠军版本被保留；挑战者失败：" + "; ".join(failed)
    return name_c, "挑战者满足了所有三个条件"

def best_of(version):
    reg = pd.read_csv(REGISTRY)
    rows = reg[(reg.version == version) & (reg.status == "ok")]
    return None if rows.empty else rows.sort_values("val_sharpe", ascending=False).iloc[0]

print(SELECTION_RULE)

现在，这个选择规则在代理程序看到任何实验结果之前就已经被确定了：

这里存在两个层次的选择过程。

best_of() 首先会使用Sharpe验证方法，在某个特定版本范围内找出表现最优秀的配置方案。但即便在内部评估中胜出，该策略也不一定会被选为最终方案。select_champion() 会将这个候选方案与当前使用的策略在所有三个评估标准上进行对比。

在这些评估标准中，特意没有包含“开发阶段的表现”。代理们可以利用开发阶段的测试结果来判断某项变更是否达到了预期效果，但如果开发阶段的性能提升幅度很大，却无法弥补验证数据不足所带来的问题。

一旦代理开始修改策略，这种区分就会变得非常重要。某个新版本在开发阶段的表现可能非常好，但最终仍可能会被拒绝采用。

建立手动基准线

在将研究工具交给Deep Agents之前，我们会先手动运行初始策略，并通过相同的评估流程进行测试。这样就能得到一个明确的参考点，同时也能确保在任何代理开始修改策略之前，数据、策略逻辑、回测引擎以及各项基准计算结果都是一致的。

这个基准线采用了126天的调整后收盘价动量指标，同时还结合了成交量筛选条件。每个月末，只有当某只ETF的动量为正值，并且其20天平均成交量高于120天平均成交量时，它才会被纳入评估范围。该策略会根据这些指标对符合条件的ETF进行排序，然后等权重持有排名前三的股票；如果没有符合条件的股票，就会保持现金持仓状态。

def manualbaseline(data, mom_window=126, vol_short=20, vol_long=120,
                    vol_ratio_min=1.0, top_n=3):
    adj, cls, vol = data["adj_close"], data["close"], data["volume"]
    mom = adj.pct_change(mom_window)
    dv = cls * vol
    ratio = dv.rolling(vol_short).mean() / dv.rolling(vol_long).mean()
    ok = (mom > 0) & (ratio > vol_ratio_min)
    dates = pd.DatetimeIndex(adj.index.to_series().resample("ME").last().dropna())
    w = pd.DataFrame(0.0, index=dates, columns=adj.columns)
    for d in dates:
        picks = mom.loc[d][ok.loc[d]].dropna().nlargest(top_n).index
        if len(picks):
            w.loc[d, picks] = 1.0 / len(picks)
    return w

d = DATA["dev"]
bt = backtest(manualbaseline(d), d["returns"])
ev = bt.loc[d["eval_start"]:]
spy = d["returns"]["SPY"].loc[d["eval_start"]:]
print(metrics(ev, spy))

fig, ax = plt.subplots(2, 1, figsize=(9, 5), sharex=True, height_ratios=[2, 1])
eq = (1 + ev["ret")).cumprod()
ax[0].plot(eq, label="strategy"); ax[0].plot((1 + spy).cumprod(), label="SPY")
ax[0].set_yscale("log"); ax[0].legend(); ax[0].set_title("Development 2005-2017")
ax[1].fill_between(eq.index, (eq / eq.cummax() - 1), 0, alpha=.4)
ax[1].set_ylabel("drawdown")
plt.tight_layout()
plt.show()

开发阶段的测试结果如下：

{'cagr': 0.0549, 'ann_ret': 0.0642, 'vol': 0.1463, 'sharpe': 0.4387, 'sortino': 0.6047, 'max_dd': -0.2606, 'ann_turnover': 11.6605, 'ann_cost': 0.0117, 'avg_cash': 0.2109, 'bench_cagr': 0.0847}

在这段发展期间，该策略的年复合收益率为5.49%，夏普比为0.4387，最大回撤幅度为-26.06%。而SPY在同一时期的年复合收益率为8.47%，因此我们有意选择采用一种回报表现相对较低的策略，而不是直接给智能代理提供已经优化好的结果。

这张股票曲线图为我们提供了更多信息：该策略在2008年市场暴跌期间避开了大部分损失，在市场恢复阶段也失去了部分优势；因此，虽然它的最大回撤幅度较低，但相应的回报也有所牺牲。

交易活动也是该策略的一个弱点。其年度交易次数为11.6605次，按照每10个基点计算交易成本的话，年交易成本约为1.17%。此外，该策略平均会将投资组合中约21.09%的资金以现金形式持有。

最重要的是，这些结果与我们之前计算的volume_mom基准指标完全一致。这一事实说明，我们手动设计的策略以及所使用的评估系统确实能够稳定地运行。

配置深度智能代理研究团队

确定性的研究环节现在已经完成。各种策略只能通过固定的评估系统进行测试，所有实验过程都会被记录下来，而且选择规则也已经明确了挑战者需要满足哪些条件才能取代当前的“冠军”策略。

现在我们可以开始添加智能代理层了。

我会将研究工作分为三个角色来完成：

• 一位策略工程师，负责实施并测试各种想法

• 一位研究评审员，负责对评估结果提出质疑

• 一位协调者，负责安排各项工作的顺序并执行选择规则

这种分工是经过深思熟虑的。同一个智能代理不应该同时具备提出策略、评估自身工作结果以及决定该策略是否应该被采纳的能力。

1. 明确智能代理的角色与职责边界

首先，我们需要初始化团队使用的各种模型：

load_dotenv(override=True)
from deepagents import create_deep_agent, FilesystemPermission
from deepagents.backends import FilesystemBackend
from langchain.chat_models import init_chat_model
from langgraph.checkpoint.memory import InMemorySaver

MODEL_ID = "openai:gpt-5.6-terra"
WORKER = init-chat_model(MODEL_ID, reasoning={"effort": "low"})
MANAGER = init_chat_model(MODEL_ID, reasoning {"effort": "medium"})

策略工程师使用较低的推理能力设置，因为他的主要工作是实施策略；而协调者和评审员则需要对比不同的评估结果、对结论提出质疑，并做出研究决策，因此他们使用较高的设定。

这些代理程序也需要一个关于“有效策略应该具备什么特征”的共同定义。我们不会让每个版本都自行设计接口，而是会为它们提供与确定性引擎所期望的相同的策略契约：

CONTRACT = """
每个策略文件都必须定义一个函数：

    def target_weights(data, **params) -> pd.DataFrame

    index: 再平衡日期，这些日期必须全部存在于 data["adj_close"].index 中
    columns: 这九个股票代码
    values: 目标权重值，每一行的数值之和必须小于等于 1.0（剩余部分表示现金）

数据键：adj_close、close、volume、returns（类型为 DataFrame，包含日期和股票代码）
使用 adj_close 来计算动量指标和回报率；使用 close * volume 来计算交易额。
日期为 t 的那一行代表在 t 日根据收盘价做出的决策；引擎会在 t+1 日执行这一决策。
需要注意避免出现空选择的情况：如果没有任何符合条件的选项，那么这一行的数值就应该设为零。

你的代码将在一个隔离的子进程中运行，该进程没有网络连接，也没有任何认证信息或保留数据。只需导入 pandas 和 numpy 即可。

基本框架如下：

import pandas as pd
def target_weights(data, mom_window=126, top_n=3):
    adj = data["adj_close"]
    mom = adj.pct_change(mom_window)
    dates = pd.DatetimeIndex(adj.index.to_series().resample("ME").last().dropna())
    w = pd.DataFrame(0.0, index=dates, columns=adj.columns)
    for d in dates:
        picks = mom.loc[d].dropna().nlargest(top_n).index
        if len(picks):
            w.loc[d, picks] = 1.0 / len(picks)
    return w
"""

这样的设计能够确保所有版本都能使用相同的评估机制进行测试。工程师可以自由改变目标权重的生成方式，但不得更改输入数据的格式，也不能绕过负责对这些权重进行评分的引擎。

接下来，我们会把前面章节中提到的研究控制措施直接纳入代理程序的指令中：

RULES = f"""
文件结构：/strategies/vN.py、/results/、/reviews/、/registry.csv、/decisions.jsonl

阶段性的限制措施，由扫描工具强制执行：
在 v(N-1) 版本成功运行、/reviews/v(N-1).md 文件中完成了相应的评审，并且通过 record_decision 函数记录下了最终决策之前，vN 版本是无法被启动进行测试的。这一点是不可违反的。

硬性限制：每个版本最多只能有 3 个版本；每个版本最多可以设置 12 种配置方案；每次修订时，结构上最多只能进行 1 次重大修改。引擎类型、数据集范围、分割方式、基准测试指标以及成本计算规则都是固定的。对于保留数据这部分内容，你们是不允许使用的；千万不要提出这样的要求。

{SELECTION_RULE}

一些基准测试指标是在任何版本编写之前就已经确定好的：
{BENCH_TEXT}

除非被告知某个特定文件确实存在并且你需要获取其内容，否则不要使用 ls、glob、grep 或 read_file 等命令。

重要的是，这些规则并不是专门为代理程序新制定的。它们实际上体现了我们之前在 Python 代码中已经设置好的各种限制条件：最多只能有 3 个版本，搜索范围是受限的，基准测试指标和成本都是固定的，存在阶段性的限制措施，而且不允许使用保留数据。

现在我们可以创建这两个专门的角色了。

策略工程师会收到这份策略契约以及 sweep() 工具。

engineer = {
    "name": "strategy-engineer",
    "description": "负责编写策略文件，并通过批量调用将这些文件提交到固定的回测系统中进行测试。适用于任何需要生成代码或统计数据的任务。",
    "systemprompt": f"""你的职责是实现策略，而非决定应该实施哪些策略。
{RULES}{CONTRACT}
操作步骤：
1. 使用 write_file 函数编写策略文件。
2. 将所有的参数组合以 JSON 格式提交一次进行测试。切勿针对每种配置分别执行测试。
3. 如果测试出现错误，请查看错误信息，修改相关文件后重新提交测试。测试失败次数会计入预算指标中。
4. 请用不超过 200 字的文字进行反馈：包括文件名称、返回的结果表格内容，以及你推荐采用的配置方案及原因。切勿直接粘贴代码。",
    "tools": [sweep],
    "model": WORKER,
}

该角色的权限被明确限制在了特定范围内：工程师可以编写策略并通过 sweep() 函数生成测试结果，但无法决定下一步应该研究什么内容，也无法自行修改现有的最优策略。

而“研究评审员”的职责则完全相反：

critic = {
    "name": "research-critic",
    "description": "负责分析测试结果表格，找出其中存在的弱点，并提出相应的改进措施。每个版本经过测试后都需进行此类评估。",
    "systemprompt": f"""你的任务是审查测试结果。你无权编写或修改任何策略代码。
{RULES}
测试结果表格会在任务描述中提供，请勿自行寻找。只有在与早期版本进行对比时，才需要使用 readRegistry 函数。
请按照以下五个标题撰写评审报告：
弱点         : 用一句话说明存在的问题
证据         : 从结果表格中找出具体的数据，并与固定基准值进行对比
改进措施     : 提出一种结构上的调整方案，并说明可能带来的过拟合风险
预期效果     : 阐明该策略应该对哪些指标产生何种影响，以及原因
过拟合风险    : 分析这种调整方案可能导致的过拟合问题，以及如何避免这种情况

tools": [readRegistry], "model": MANAGER, "permissions": [ FilesystemPermission(operations=["write"], paths=["/strategies/**"], mode="deny"), FilesystemPermission(operations=["read","write"], paths ["/**"], mode="allow"), ], }

评审员的任务不仅仅是判断某个策略“看起来是否不错”；他们必须找出该策略的弱点，并用具体证据支持这一结论，同时提出可行的改进措施，并明确说明这些措施可能带来的过拟合风险。

更重要的是，这种权限划分在系统层面得到了严格执行：评审员被明确禁止对 /strategies/** 目录进行写入操作。他们可以查看研究结果并撰写评审报告，但无法擅自修改那些需要评估的策略文件。

2. 创建协调者

协调者将工程师和评审专家连接起来，形成一个完整的研究流程。

COORDINATOR = f"""你负责执行定量研究流程，评估标准在于该流程的透明度，而非最终结果。
{RULES}
对于每个版本N，你需要按照以下步骤操作：
1. 使用write_todos制定计划；
2. 将实施细节委托给策略工程师处理；
3. 将工程师整理好的数据原封不动地传递给评审专家；
4> 自己根据规则进行筛选，并说明哪些环节通过或失败；
5> 调用record_decision函数，提交最终结果及你的判断理由。

步骤5是必须完成的。在下一个版本完成之前，当前版本将处于等待状态。
切勿将仅仅属于参数调整的内容伪装成结构化的研究方案。即使出现了新版本，也不得随意替换旧版本。"

协调者负责管理整个研究流程，但它仍然建立在我们已经建立的确定性控制机制之上。它无法使工程师报告的夏普比率具有正式效力，也无法绕过实验注册系统或违反固定规则来推进某个策略的研究。

文件系统后端为团队提供了用于存储策略文件、结果、评审记录和决策内容的共享工作空间。virtual_mode=True这一设置使得该工作空间可以通过诸如/strategies/v1.py这样的路径被访问，而实际上这些文件都存储在底层的实际研究目录中。

我们还会将整个v1 -> v2 -> v3的研究流程放在一个受检查点保护的线程中执行，并使用一个小辅助函数来调用协调者：

def run(prompt):
    out = agent.invoke({"messages": [{"role":"user","content":prompt}]}, THREAD)
    c = out["messages"][-1].content
    print(c if isinstance(c, str) else
          "\n".join(b.get("text","") for b in c if b.get("type") == "text"))
    return out

print("子代理模型：", engineer["model"].model_name, critic["model"].model_name)
print(WORKER.invoke("回复‘ok’").content)

最后的检查确认了所有子代理模型都已成功初始化：

子代理模型：gpt-5.6-terra gpt-5.6-terra
[{'type': 'text', 'text': 'ok', 'annotations': [], 'id': 'msg_09ea14bfb753e624006a72189dbf84819eac295e52e7d7ccd0', 'phase': 'final_answer'}]

此时，研究团队已经拥有了进行工作所需的一切条件：工程师可以实施并测试各种策略，评审专家可以在不修改代码的情况下对相关数据提出质疑，而协调者则只有在每个版本都经过测试、评审并获得正式决定后，才能继续推进后续的研究工作。

复现版本v1的手动基准策略

第一个代理周期不应引入新的策略思路。我们已经有了经过人工验证的基准策略，因此v1为我们提供了一种可控的方法，用于检验新的代理工作流程是否能够再现这一策略、运行预定义的实验、获取独立的评审意见，并在开始任何实质性修改之前记录下相关决策。

该基准策略的逻辑在任务要求中已有明确规定。唯一需要进行的搜索是在三个动量窗口和三种投资组合规模范围内，对九种配置进行预先设定的遍历测试；这种设计可以防止工程师在看到初步结果后就扩大参数搜索的范围。

V1_BRIEF = """版本1的基准策略。

具体实现由策略工程师负责。/strategies/v1.py：使用调整后的收盘价计算126天的动量指标；使用原始收盘价和原始交易量计算过去120天内20天的日均交易量；只有当动量指标大于0且交易量比率大于1.0时，该配置才符合条件；根据动量指标对候选配置进行排序，排名前三的配置按相同权重进行投资，每月重新平衡投资组合，否则将资金转为现金。相关参数包括：mom_window、vol_short、vol_long、vol_ratio_min、top_n。

只需在一次调用中遍历这九种配置：将mom_window设置为(63, 126, 189)中的任意值，并与top_n设置为(2, 3, 4)中的任意值，其余参数均使用默认值。

将生成的配置列表传递给评审专家，让其与固定的基准指标进行对比评估。然后应用选择规则（由于V1没有现有的竞争方案，因此它将默认被选为最优方案），并调用record_decision函数来记录最终决策结果。

最后需要报告的内容包括：所选择的配置方案、其与等权重持有策略及纯动量策略的比较结果，以及评审专家提出的修改建议与你的最终决定。”

完整的代理周期运行结束后会返回以下结果：

## 版本1的基准策略测试完成

**最优方案：** v1（初始版本；由于没有竞争方案，因此默认被选为最优方案）  
**所选择的配置参数如下：**
```json
{
  "mom_window": 126,
  "vol_short": 20,
  "vol_long": 120,
  "vol_ratio_min": 1.0,
  "top_n": 2
}
```

选择这个配置是因为在九种预定义的配置方案中，它的验证阶段Sharpe比率最高，为**0.542**。

### 各方案对比结果

| 指标              | V1          | 等权重持有策略    | 纯动量策略      |
|-----------------|-------------|--------------|--------------|
| **发展阶段的年化复合增长率** | 0.0550       | 0.0904       | 0.0750       |
| **发展阶段的Sharpe比率** | 0.4240       | 0.5532       | 0.5363       |
| **发展阶段的最大回撤幅度** | -0.2760       | -0.5203       | -0.2817       |
| **发展阶段的年交易量** | 11.5890      | 0.0000       | 7.2798       |
| **验证阶段的年化复合增长率** | 0.1000       | 0.1769       | 0.2051       |
| **验证阶段的Sharpe比率** | 0.5420       | 0.8697       | 0.9279       |
| **验证阶段的最大回撤幅度** | -0.2950       | -0.3371       | -0.2901       |

V1的发展阶段最大回撤幅度低于另外两种策略，但在发展阶段的年化复合增长率和Sharpe比率方面表现都不如它们。此外，V1的年交易量也远高于纯动量策略（V1的年交易量为**11.5890**，而纯动量策略为**7.2798**）。

### 选择规则的应用

由于V1**没有现有的竞争方案**，因此它默认被选为最优方案。

- 第一筛选标准——验证阶段的Sharpe比率：不适用
- 第二筛选标准——验证阶段的最大回撤幅度：不适用
- 第三筛选标准——发展阶段的年交易量：不适用

最终决定将V1作为基准方案保留。

工程师完成了全部九次测试，并选择了验证Sharpe值最高的配置：

{
  "mom_window": 126,
  "vol_short": 20,
  "vol_long": 120,
  "vol_ratio_min": 1.0,
  "top_n": 2
}

该配置的验证Sharpe值为0.542。在v1版本的测试中，这是表现最好的配置；不过，由于有固定的评估标准，我们才能避免将“本次测试中的最佳方案”与“真正的优秀策略”混为一谈。

在发展阶段的年均增长率和验证阶段的Sharpe值方面，v1版本仍然不如等权重买入并持有策略以及单纯的动量投资策略。此外，v1版本的交易频率也远高于单纯动量投资策略。虽然该策略在发展阶段的回撤幅度较小，但这一优势本身并不足以使其整体表现更加出色。

由于目前还没有其他竞争者出现，因此三个晋级标准并不适用。v1版本自然而然地成为了后续所有版本都需要超越的初始“冠军”。

接下来，评估专家开始关注其他方案。那些周期为189天的方案在发展阶段的表现更好，但这一优势在验证阶段就被削弱了；而周期为126天的方案在不同投资组合规模下表现更为稳定，不过它们的验证Sharpe值仍然远低于那些简单的基准指标。

评估专家并没有建议调整动量窗口的长度或top_n的值，而是提出了一种结构上的改进：加入一个针对整个市场的绝对动量筛选机制。当SPY指数的动量值为正时，现有的动量投资组合会继续保持活跃状态；而当市场趋势转为负向时，该组合就会转换为现金。

在继续下一步之前，我们可以确认：整个v1版本的测试过程确实满足了阶段评估所要求的三个条件：完成了成功的实验、经过了专家的评审，并且决策过程也被记录了下来。

print(pd.read_csv(REGISTRY).groupby(["version","status")).size())
print("决策记录:", [d["version"] for d in _decisions:])
assert (WS/"reviews"/"v1.md").exists(), "缺少v1版本的评估报告"
assert any(d["version"] == "v1" for d in _decisions()), "缺少关于v1版本的决策记录"
print("v1版本测试流程已完成")

第一阶段的测试已经完成。v1版本现在成为了当前的“冠军”；评估专家也指出了它存在的具体弱点，因此下一个版本的开发将围绕这一弱点来进行。

让代理们来修改策略吧

既然v1版本已经被确立为基准“冠军”，那么研究流程就可以从单纯的复制阶段进入改进阶段了。从现在开始，每一个新版本的开发都必须针对在上一次评估中发现的弱点来进行；只有当一个候选方案能够通过所有三个固定的筛选标准时，它才能取代现有的“冠军”版本。

在v2版本中测试市场趋势筛选机制

v1版本的评审者提出了一项结构上的修改：保留现有的横截面动量策略，但当SPY的绝对动量为负值时，将投资组合转换为现金。

我们不会立即重新开始参数搜索，而是会使用当前的v1版本配置来测试这一修改。因此，第一次v2版本的对比主要关注的是机制筛选器本身，而不是另一轮参数调整。

V2_BRIEF = """请查阅/reviews/v1.md中的评审者v1版本提案。如果您同意该方案，请让strategy-engineer将其实现为/strategies/v2.py文件。不要修改v1.py文件。每次调用最多测试12种配置。将结果表格传递给research-critic，由其撰写评论并保存至/reviews/v2.md文件。之后根据选择规则在v2版本与当前版本之间进行对比，记录下三种判断标准中哪些通过了检验，哪些没有通过，并调用record_decision函数为v2版本生成最终决策。不要继续执行后续步骤。"""

_ = run(V2_BRIEF)

代理团队给出的回复是：

同意将v1版本的评审者提案作为一项结构上的修改来实施：即采用双动量机制筛选器，当SPY在当前`mom_window`时间窗口内的调整后收盘价动量为负值时，将投资组合转换为现金。

- 实现方式：/strategies/v2.py
- /strategies/v1.py文件未作任何修改。
- 使用当前的v1版本参数测试了1种配置：

```json
{
  "mom_window": 126,
  "vol_short": 20,
  "vol_long": 120,
  "vol_ratio_min": 1.0,
  "top_n": 2
}
```

### V2版本的结果

| 指标          | V1版本       | V2版本     |
|---------------|-------------|------------|
| 发展阶段夏普比率 | 0.424        | 0.611      |
| 发展阶段最大回撤率 | -0.276       | -0.190      |
| 发展阶段年周转率   | 11.589      | 9.814      |
| 验证阶段夏普比率 | 0.542        | 0.321      |
| 验证阶段最大回撤率 | -0.295       | -0.295      |

评审者的评论记录在/reviews/v2.md文件中。

### 选择规则：v2版本与v1版本的对比

1. **验证阶段夏普比率——未通过**  
   V2版本：0.321，V1版本：0.542。

2. **验证阶段最大回撤率——通过**  
   V2版本：-0.295，V1版本：-0.295；两者之间的差异仅为0.0个百分点，仍在允许的2个百分点范围内。

3. **发展阶段年周转率——通过**  
   V2版本：9.814，V1版本：11.589；V2版本的年周转率低于V1版本，也低于允许的最高上限13.907%。

由于v2版本在验证阶段夏普比率这一指标上未通过评估，因此它**不会**取代当前的优选策略。最终决策记录为：**继续使用v1版本作为优选策略**。

乍一看，v2版本似乎是一项重大的改进。发展阶段的夏普比率从0.4235提高到了0.6110，最大回撤率也从-27.57%降低到了-18.99%，年周转率则从11.5888下降到了9.8139。

如果只看发展阶段的数据，这种机制筛选器似乎一下子解决了多个问题。

但验证阶段的测试结果却截然不同。v1版本的夏普比率从0.5424降到了v2版本的0.3207，而最大回撤率基本上没有变化。因此，发展阶段取得的这些改进在决定该策略是否能够被继续采用时，并没有起到预期的作用。

正是在这里，选择规则发挥了它的作用。V2通过了回撤率测试，也轻松满足了周转率要求，但它未能满足第一个条件：其夏普比率验证结果不能比现有的方案更差。

因此，尽管V2在发展效果上表现得更好，它仍然无法取代现有的冠军方案。

评论者还指出了这些证据中的另一个缺陷：V2仅在一个配置条件下进行了测试，这意味着其显著的改进并没有得到其他相关参数的支持。评论者建议不对规则本身进行调整，而是对系统结构进行重新设计：用基于波动性调整的权重来替代现有的二进制美元成交量筛选机制，从而对选定的动量资产进行加权处理。 在测试这个方案之前，我们首先需要确保与V2相关的实验数据、评审记录以及最终决策结果都已经被妥善保存下来。

因此，V2为我们提供了有用的数据，但却没有获得晋升的机会。

在v3版本中实施最终修订

评论者提出的建议被采纳为最终的修订方案。v3版本会保留v2中引入的广泛市场筛选机制，同时取消二进制美元成交量限制规则，并根据资产近期波动性的大小来调整其权重。 这一次，工程师将测试三种不同规模的投资组合组合，在这些组合中，top_n的值分别被设置为2、3和4。在经过最终的评审和决策流程后，协调者必须立即冻结那些仍然符合冠军标准的投资策略。

完整的执行步骤如下：

V3_BRIEF = """按照最终确定的修订方案，将代码保存为/v3.py。不要修改v1或v2版本。一次测试最多可以覆盖12种配置组合，在/reviews/v3.md中获取评论者的反馈意见，然后应用选择规则，并记录最终的决策结果。最后将结果写入/frozen.json文件，文件内容应包含以下信息： {"version": "冠军方案版本", "params": {}, "rationale": "..."} 其中，"version"字段应填写选择规则所认定的冠军方案版本，这个版本可能是v1、v2或v3。完成这些操作后，即可结束测试过程。""" _ = run(V3_BRIEF) display(Markdown("### 决策记录")) for dd_ in _decisions(): print(f"{dd_['version']} -> 冠军方案：{dd_['champion']}: {dd_['rationale'][:160]}") print("\nfrozen方案信息：", (WS/"strategies"/"frozen.json").read_text())

最终的测试结果如下：

最强的v3配置使用了top_n=3这一参数，其验证阶段的夏普比率达到了0.5377。这个数值与v1版本的0.5424非常接近。此外，v3还将验证阶段的回撤幅度从-0.2954降低到了-0.2884，同时也将开发阶段的周转率从11.5888降到了7.0480。

因此，在这三个评估标准中，有两个都是满足的。

验证阶段夏普比率之间的差异仅为0.0047，这一点使得这个决策成为整个实验中最关键的环节之一。虽然这些数值实际上几乎完全相同，但考虑到v3在回撤幅度和周转率方面的表现更优，选择v3似乎也是理所应当的。

然而，这样做就意味着在看到实验结果之后才改变原有的评判标准。

这个评判规则是在v3出现之前就已经确定的，它要求新版本的夏普比率不得低于现有版本。而v3虽然勉强满足了这一要求，但差距确实很小。

因此，v1最终还是被确定为胜者。

协调员会将这一结果保存到frozen.json文件中，其中会包含所有在整个研究过程中被证明有效的参数设置。至此，策略选择阶段就已经结束。此后发生的任何事情都不允许改变哪个版本最终能够进入测试阶段。

确定胜者并开启测试阶段

研究流程已经完成，但尚未公布最终的胜者方案。在正式公开之前，我们需要确认三个策略测试周期都已经结束，并且胜者版本已经被确定下来。

这一检查是在主研究流程之外的地方进行的。这一点非常重要——如果是由代理程序本身来决定何时公布胜者方案，那么这个时间点就会受到代理程序行为的影响，而而不是由整个系统来决定的。

frozen = json.loads((WS/"strategies"/"frozen.json").read_text())
print("frozen:", frozen)
assert len(_decisions()) == 3, f"预期有3个决策结果，实际得到{len(_decisions())}"
for v in ["v1","v2","v3"]:
    assert (WS/"reviews"/f"{v}.md").exists(), f"缺少关于{v}的评估报告"
    assert not pd.read_csv(REGISTRY).query(f"version=='{v}' and status=='ok'").empty, f>版本{v}没有测试记录
print("三个测试周期都已经完成")

for field in ["adj_close","close","volume"]:
    DATA["holdout"][field].to_parquet(WS/"data"/f"holdout_{field}.parquet")

final = {}
for split in ["dev","val","holdout":
    res = run_isolated(WS/"strategies"/f"{frozen['version']}.py", frozen["params"], split)
    assert res["ok"], res["error"]
    final[split] = res["metrics"]
    plt.plot(pd.Series(res["equity"], index=pd.to_datetime(res["dates"])), label=split)
plt.yscale("log"); plt.legend(); plt.title(f"所有时期的frozen {frozen['version']}版本表现")
plt.show()

(WS/"results"/"holdout.json").write_text(json.dumps(final, indent=2))
BENCH_HOLD = benchmark_table("holdout")
comparison = pd.concat([pd.DataFrame(final).T.assign(source="strategy"),
                        BENCH_HOLD.assign(source="benchmark_holdout")])
comparison[["cagr","sharpe","sortino","max_dd","ann_turnover","source"]}]

这些检查确认，在进行保留测试之前选定的v1配置仍然保持不变：

frozen: {
    'version': 'v1',
    'params': {
        'mom_window': 126,
        'vol_short': 20,
        'vol_long': 120,
        'vol_ratio_min': 1.0,
        'top_n': 2
    },
    'rationale': "在v3未能通过所需的夏普比率验证标准（v3的夏普比率为0.538，而v1为0.542）之后，v1依然被认为是最佳策略；尽管v3通过了回撤幅度和收益波动性相关的验证标准。"
}
所有三个测试周期均已完成

只有在这些检查都通过后，工作流程才会将保留测试数据提供给评估人员，并开始对这种被冻结的策略进行评估。

该图表显示，在开发阶段、验证阶段以及保留测试阶段，始终采用的是相同的v1配置。

每个测试阶段都是独立进行评估的，因此这三条曲线不应被理解为代表一个连续复合投资的组合表现。关键在于：在所有这三个阶段中，该策略的逻辑和参数都保持不变。

最后的对比结果如下：

在未参与保留测试的阶段，采用v1配置所获得的年化复合收益率为13.98%，夏普比率为0.7962；这两个数值都高于在同一时期采用SPY指数进行买入并持有操作、等权重投资策略、单纯基于动量因素的投资策略，以及基于成交量和动量因素的组合投资策略所取得的成绩。

其最大回撤幅度为-23.04%，这一数值也略低于采用SPY指数进行投资或单纯基于动量因素的投资策略所取得的最大回撤幅度-18.23%。

这个结果确实相当不错，但它并没有改变我们在进行保留测试之前所得出的结论：v1配置在验证阶段的表现仍然不如那些更简单的基准策略；而且，在这些数值被确定下来之前，我们就已经决定将v1配置冻结起来了。

保留测试为我们提供了对该预先确定的策略的一次未经公开评估的机会；但它并没有给我们第二次机会来选择究竟应该测试哪种投资策略。

审核整个研究过程

在结束这项实验之前，我们会给协调者布置一项最后的任务：在所有数据都被冻结之后，重新审查整个研究过程。

在这个阶段，任何结果都已经无法改变原本选定的投资策略了。协调者会收到被冻结的配置参数、三个测试阶段的各项指标数据、保留测试阶段的基准结果、实验记录信息、决策过程的相关记录，以及其他专家的意见评论。我还会明确要求协调者不要为最终的结果进行辩护。

REPORT_BRIEF = f"""保留测试已经完成一次，相应的投资策略也已被冻结。现在任何事情都无法再改变了。

Frozen: {json.dumps(frozen)}
各测试阶段的指标数据：{json.dumps(final)}
保留测试阶段的基准结果：{BENCH_HOLD[COLS_B].to_json()}

请执行以下操作：
1. 无参数地调用readregistry函数；
2. 读取/decisions.jsonl文件以及/reviews/目录下的所有文件；
3. 编写/report.md文件，内容应包括：
   - 每个版本中发生了哪些变化，是什么因素导致了这些变化；
   - 选择机制是如何确定最终的最佳策略的，其中哪些评估标准未能通过；
   - 这些修改是否改善了研究结论，这一点需要与投资回报分开来看；
   - 被冻结后的投资策略在保留测试阶段与SPY指数进行买入并持有操作、等权重投资策略、单纯基于动量因素的投资策略相比表现如何；
   - 基于成交量筛选机制是否真的能够带来预期的收益波动性；
   - 在哪些决策过程中，你可能考虑不周、依据的证据不够充分，或者只是运气好而已。

请在报告中引用注册系统中记录的实验编号。不要为最终的结果进行辩护。

让我们将这份报告与完整的实验记录一起展示出来，确认每一个版本都对应有相应的运行结果、决策内容以及评审意见：

display(Markdown("## 代理报告"))
display(Markdown((WS / "report.md").read_text(encoding="utf-8")))

display(Markdown("## 实验记录")
reg = pd.read_csv(REGISTRY)
display(reg[["version","run","status","params","dev_sharpe","dev_sortino",
             "dev_max_dd","dev_turnover","val_sharpe","val_max-dd","dev_cagr_20bps"]])
print("包含运行记录的版本：", sorted(reg["version"].unique()))
print("被记录下来的决策结果：", [d["version"] for d in _decisions()])
print("磁盘上的评审文件列表：  ", sorted(p.stem for p in (WS/"reviews").glob("*.md")))

这种审计机制对于了解研究过程的进行方式而言更为有用，而非用于进一步的性能比较。

它揭示了三个明显的缺陷：V2版本仅在一种配置下进行了重大的设计变更，因此其改进效果的可靠性缺乏足够的证据支持；而V3版本与原始的v1版本相比存在多处差异，这使得人们难以确定究竟是哪些因素导致了它的行为变化。

更重要的是，这次审计还发现评审机制本身存在错误。V3版本的评审建议用波动率缩放来替代原有的二进制体积比率过滤器，但实际上V3版本已经去掉了那个过滤器，并采用了逆波动率加权方法。这个解释听起来似乎合理，但并没有准确反映所评估的策略内容。

这或许是从这次审计中得到的最重要启示：虽然按照不同角色来划分代理是有用的，但这并不能保证这些代理真正理解它们所评估的对象。保留策略代码、实验记录、评审意见以及决策过程，可以为我们提供独立的验证依据，从而检验他们的推理是否正确。

结论

最终，我们的构建工作已经完成。

我们最初使用的是原始的EODHD市场数据，而最终得到了一个结构完备的多代理研究系统：该系统拥有固定的数据边界、确定性的回测工具、基准测试机制、实验跟踪功能，同时包含了三种代理角色、三个策略版本，以及一系列的测试环节。

不过，整个开发过程远没有“人工智能不断优化策略”这种说法看起来那么简单顺利。V2版本在开发阶段表现不错，但在验证环节却失败了；V3版本的绩效仅比v1版本低0.0047的夏普比率。甚至评审机制也误解了它所评估的策略内容。

奇怪的是，正是这些复杂的问题使得这项实验变得有意义——它们清楚地说明了为什么对代理的行为进行控制是如此重要。

目前仍有许多地方需要进一步完善，比如加强稳定性测试、改进单一变更的影响分析方法，以及设立独立的评审机制和参数稳定性检测流程。<但关键在于：智能助手在帮助生成研究思路以及对这些思路进行评估方面确实能够发挥重要作用。只不过，它们不应该被允许控制那些决定这些思路能否被采纳的证据。