DEV Community: Henry Lin

RD-AGENT 第六讲 · 一次预注册实验的完整生命周期：两份判决书

Henry Lin — Wed, 14 Oct 2026 05:06:06 +0000

第六讲 · 一次预注册实验的完整生命周期：两份判决书

第二章 · 诚实的验证（第三讲，本章收尾）
学习目标：跟踪两个真实假设从立项到判决的全过程，掌握预注册实验的操作细节与三种判决的分界。
前置：第四、五讲。

1. 为什么选这两个案例

第五讲的全 FAIL 是"批量候选"的结局；本讲看单个假设的完整生命周期——人提出想法、写成预注册、走完验证、拿到判决、执行判决后果。两个案例一个 REJECTED、一个 INCONCLUSIVE，正好覆盖"否决后怎么办"和"证据不足时怎么办"这两个最难的操作。（SUPPORTED 的案例留给第四章，避免结构上"大团圆"得太早。）

2. 案例一：业绩预告事件窗 —— REJECTED

2.1 假设

A 股业绩预告披露前后存在可交易的定价偏差窗口：预告超预期的股票在事件后若干日仍有漂移（盈余公告后漂移，PEAD 的预告版）。数据基础：业绩预告数据（forecast）+ 曾用名变更（namechange）面板。

2.2 预注册要点

判决前锁定的要素（节选）：事件窗定义、对照组构造、OOS 窗口、价差门槛（方向 + 量级 + 显著性）、以及三种判决的边界。在触碰任何 OOS 数据之前完成存档。

2.3 判决

OOS 价差 −0.49%，方向反。REJECTED，一次判定。

发现期（IS 段）同样没有可用的正价差——这个细节很重要：它不是"发现期有效、样本外翻车"的经典过拟合形态，而是从头到尾就不存在。假设本身错了，不是数据 unlucky。

2.4 判决的执行——最见制度成色的部分

REJECTED 落地后发生了三件事，每件都值得讲：

假设族封禁：事件窗这一族假设进入"禁止复测"清单——不允许换个窗口定义、换个对照构造再申请一次 OOS。同一 OOS 窗口是消耗品，已经在这件事上花掉了。
数据资产保留：forecast + namechange 面板入库为可复用资产。否决的是假设，不是数据——数据没有立场，以后别的假设还能用。
衍生观察降级：过程中注意到"预告后反转"的苗头，但因未预注册、且属假设族内变体，标记为 hypothesis-generating（仅产生假设，不构成证据），禁止直接进入下一轮验证。想追，先写新预注册。

第 3 条最微妙，也最常被破坏："我在验证 A 的时候顺便看见 B 好像行"——这个"看见"本身已经消耗了数据的一次查看权。制度上的正确处理不是假装没看见，而是把它降级为假设、重新走全流程。顺便看见的东西不是证据，是线索。

3. 案例二：指数调样流 —— INCONCLUSIVE

3.1 假设

指数定期调样时，被动资金按公告名单集中买卖，造成可预测的短期价格压力（调样公告日到生效日之间的流入/流出流）。这是文献中有先例的现象（index rebalancing effect），先验等级比 AI 生成的叠加因子高一档（第三讲的结论二在此适用）。

3.2 流程与装配验证

这个案例多了一个真实工程环节：装配检验。调样流策略的输入是"公告名单 → 被动规模估算 → 流向符号"的拼装管线，任何一环错位（日期、方向、权重）都会无声地产出垃圾。装配管线先做了 PASS 检验（拼装结果与公告原文抽样比对），然后才进预注册。先验证数据管道，再验证策略假设——顺序不可反。

3.3 判决

发现期 +0.95%；OOS +1.00%，p = 0.062。INCONCLUSIVE——功效不足，冻结。

3.4 INCONCLUSIVE 的正确处理（比 REJECTED 难）

REJECTED 痛快，INCONCLUSIVE 折磨人——差 0.038 的 p 值，人性 screaming"再试一次"。制度规定：

冻结，不许换口径重测：把窗口延长一年、换个统计检验、剔除某个月——每一个都是"看完结果再调规则"，正是第四讲锁死的研究者自由度
不许宣布"基本有效"：对外口径只有三个词，INCONCLUSIVE 意味着不能写进任何汇报的"有效策略"栏
唯一的两条出路：等新的前向数据积累（时间换功效），或写新预注册走前向轮（在下一轮真实调样发生前登记好门槛，用尚未发生的数据判定）

案例的实际处置：OOS 冻结，后续按前向路线重新预注册（2026-12 的调样轮在前向监控覆盖范围内）。"等"也是判决的一部分——接受"这个问题现在无法回答"，是多检验纪律里最反人性也最省钱的一条。

4. 三种判决的决策树（本章收束）

把第四讲到本讲的全部案例叠成一张裁决速查：

预注册验证结果
├─ 过全部门槛（DSR/副判据/分年/净费/正交）
│    → SUPPORTED → 影子监控 30 天（第四章）→ 部署决策
├─ 方向反 / 分年结构违规 / 正交后无残差
│    → REJECTED → 假设族封禁，数据保留，衍生观察降级
└─ 方向对但功效不足（p 略高、窗口短）
     → INCONCLUSIVE → 冻结：等前向数据 or 新预注册前向轮
                          （禁止：改门槛、改口径、原窗重测）

注意树里没有一条边通往"差不多就先小仓位上"。仓位不是验证的替代品——用真钱试探一个未判决的假设，等于用最大的成本做一次不可复现的实验。

5. 第二章总结

独立验证 = 制度而非算力：预注册锁死自由度，DSR 校正选拔基数，OOS 一次性判定
批量候选的结局（第五讲）：四层审计栈全灭，AI 工厂的合格产出是候选假设流
单假设的结局（本讲）：REJECTED 封族留数据、INCONCLUSIVE 冻结等前向——判决的价值在执行
唯一活下来的案例年化只有 2.37%：真 alpha 又小又难看；假 alpha 又大又合理

第二章证明了"诚实的验证"是可制度化执行的。第三章转向它的地基：当数据本身有问题时，再严格的验证也会被静默污染——双重复权、基准伪影、日期戳错位，五个真实的数据暗雷。

RD-AGENT 第五讲 · 三候选 DSR 全 FAIL 复盘：超额从哪来，又到哪去了

Henry Lin — Wed, 14 Oct 2026 05:05:40 +0000

第五讲 · 三候选 DSR 全 FAIL 复盘：超额从哪来，又到哪去了

第二章 · 诚实的验证（第二讲）
学习目标：通过 A 股战役的完整复盘，看懂"排行榜很好看 → 独立验证全灭"的具体机制。
前置：第四讲。

1. 案例背景

A 股战役（2026-09-18~22）结束后，从 290 个因子的排行榜里，按预注册流程选出三个头部候选，送入独立验证管线。判定结果：

三候选 DSR 全部 FAIL。

这一讲的任务是把"全 FAIL"三个字拆开：每个候选的超额看起来从哪来、实际从哪来、验证时死在哪一刀下。

2. 第一刀：DSR——你试了多少次，就要还多少债

回顾第四讲：DSR 对 Sharpe 按试验次数 N 做校正。三个候选来自 290 个因子的排行榜头部，N 不是 3，是 290（严格说还要算上每个因子的多次评估——第二讲看过，momentum_20d 一个因子就有 23 次评估，全池评估总次数以千计）。

直觉计算：即便候选的原始 Sharpe 在"单次检验"意义下显著，经过千次级别的试验基数校正后，所需门槛被推高到常规数字几乎不可能达到的水平。这不是门槛苛刻，是事实如此——你在千次抛硬币里挑出的最好结果，本来就不该按单次抛硬币的标准评价。

三个候选全部倒在这一关。注意这一刀的kill 逻辑：它不是说"这三个因子必然无效"，而是说"从千次选拔里胜出的这三个，其成绩不足以证明自己不是运气"。证据不足，按无罪推定的反面处理：不予采信。

3. 第二刀：归因分解——超额是 alpha 还是风格 beta

验证管线对候选做的第二件事：正交化检验（对已知风格因子回归，看残差）。战役中的组合级归因结果更极端：

等权合成组合 OOS IC 0.043，看似有效；但对行业 + 低波 + 流动性三个已知风格做正交后，92% 的收益被解释掉，残差 alpha 的 t 值 = 0.14。

翻译成人话：这个"AI 因子组合"赚的钱，九成二可以由"买了一堆低波动、高流动性股票、行业分布偏斜"复现——这三样都是几十年前就知道的东西，不需要 AI 一晚上算几百个因子。真正的残差部分，t=0.14，统计上就是零。

这是排行榜最惯用的魔术：把风格 beta 包装成 alpha 卖给你。排行榜不管你赚的是不是新钱——超额收益一列不区分"发现了别人不知道的东西"和"重仓了一个已知风格"。而后者你直接买低波 ETF 就有，不需要付因子挖掘的成本。

4. 第三刀：换手与成本——毛收益买不回净收益

第三讲埋过线：33 个版本的动量类因子是换手重灾区。验证管线在净费口径下复算：头部候选的换手率水平（战役实测约 40%+ 双边/调仓期）叠加 A 股交易成本后，毛超额被侵蚀到净口径不再显著。

排行榜展示的是毛收益或半净口径（第二讲的"三个必问"之二）。工厂没有动机展示净数字——它不知道你的费率，更因为净数字会让排行榜难看。

5. 最深的一刀：基准伪影（预告第八讲）

三候选验证期间还发现了一个更基础的问题：基准本身有口径缺陷（沪深 300 口径下剔除了涨停股），导致"超额"中有一块是基准定义制造的伪影。换成诚实口径重算后，双周 top200 组合没有任何超额。

这个发现的震撼在于它的层级：它不在"因子好不好"这一层，而在"超额这个量本身可不可信"这一层。因子验证做得再严格，基准错了，全部结论的地基就错了。完整展开留给第八讲——这里只需要记住教训的形状：验证要向下问到数据的地基，而不是停在因子这一层。

6. 复盘：四刀的层级结构

把四刀叠起来看，它们其实是一个自下而上的审计栈：

第 4 层  统计充分性   DSR / 多重检验校正     —— 成绩配得上选拔基数吗
第 3 层  收益来源     正交化 / 归因分解      —— 是新钱还是旧 beta
第 2 层  可实现性     换手 × 成本敏感性      —— 净费后还在吗
第 1 层  数据地基     基准口径 / 面板构建    —— 量本身可不可信

工厂的排行榜只在第 4 层之下活动，而且只做了一半（没有校正）。独立验证的价值就是把这个栈走完——三候选在四层上各挨了一刀，没有一个是"因子逻辑本身被证伪"，全部是证据与口径不合格。这正是第四讲那句话的实例：验证判的是"是不是真的"，不是"有多好"。

7. 军心稳定问题：全 FAIL 之后怎么办

全灭之后自然的情绪反应有两种，都要抵制：

"验证太苛刻，放宽门槛"——门槛放松 = 重新引入多重检验，全 FAIL 的结论本来就是"证据不足"，放宽标准不会让证据变足，只会让假阳性进门。
"AI 挖因子没用，弃了"——过度结论。正确读法：AI 工厂的价值主张需要重述。它的合格产出不是"可直接上线的因子"，而是"低成本生成的候选假设流"。从候选流到可部署策略之间，隔着一整条验证管线——这条管线以前也存在（人类研究员的 peer review + 实盘小仓试错），只是没人把它当成生产环节的一部分来正式建设。

战役的实际决策是第三条路：保留工厂、正式建设验证制度、把"诚实口径下零超额"如实记档——然后转向。转向后的第一个预注册实验（第六讲）和唯一存活的策略（第四章），都诞生在这套制度建立之后。

8. 本讲小结

三候选死因四层级：DSR（千次选拔基数）、风格 beta 冒充 alpha（92% 被解释，残差 t=0.14）、净费侵蚀、基准伪影
排行榜的魔术：不区分新钱与旧 beta、展示毛收益、不做选拔校正
全 FAIL 的正确解读：AI 工厂的产出单位是"候选假设流"，不是"可上线因子"
审计栈四层：统计充分性 → 收益来源 → 可实现性 → 数据地基，缺一层都可能漏杀

思考题：假如某候选通过了 DSR 和正交化，但净费后超额恰好为零（毛超额 = 成本）。你会判 SUPPORTED、REJECTED 还是 INCONCLUSIVE？给出理由。（没有标准答案——真实战役的裁决是按预注册里的门槛表逐项对照，想清楚"你的门槛表里有没有这一格"才是本题考点。）

RD-AGENT 第四讲 · 为什么需要独立验证：预注册与验证管线

Henry Lin — Wed, 14 Oct 2026 05:05:14 +0000

第四讲 · 为什么需要独立验证：预注册与验证管线

第二章 · 诚实的验证（第一讲）
学习目标：理解"工厂自报成绩"与"独立验证"的制度差异；掌握预注册、DSR、样本外一次性判定等核心概念。
前置：第一章（尤其第二讲的"三个必问"）。

1. 从一个问题开始

第一章末尾留了一个数字：290 个因子，自报超额最高 +26.5%，独立验证零存活。这一讲回答："独立验证"到底独立在哪，凭什么它的结论比工厂的数字更可信？

先看一个对照。工厂回测和独立验证做的是相似的事——都是拿历史数据算因子收益。差别不在算力、不在代码质量，在制度：

维度	工厂内置回测	独立验证管线
数据	战役面板（同一份）	独立拉取、独立清洗的复算面板
规则	可以边看结果边调（循环天然如此）	预注册：判定规则先写死、先存档
判定	排行榜排个序，没有门槛	一次性判定：门槛预注册，触发即终审
对象	每个因子都出数字	只验证通过初筛的极少数候选
结论性质	描述性的（"它收益多少"）	判决性的（"拒绝/不可判定/支持"）

一句话：工厂回测回答"如果用它会怎样"，独立验证回答"它是不是真的"。前者是描述，后者是统计判决。

2. 多重检验：为什么"边看边调"必然产出假阳性

这是全部验证科学的第一性原理，值得慢慢讲透。

设 290 个因子全部是纯噪声（与未来收益零相关）。每个因子在 5% 显著性水平下做检验，期望产出 290 × 0.05 ≈ 14.5 个"显著有效"。这 14.5 个因子在排行榜上与真因子长得一模一样——显著性不会在你脸上写"我是运气"。

人手动做研究时这个问题就存在（学术界管它叫 p-hacking 的温床）；LLM 循环把基数放大了两个数量级，把成本降到了零——于是多重检验从"需要警惕的偏差"升级成"系统的主导性失效模式"。

对付它的唯一制度化手段是预注册（pre-registration）：

在看数据之前，把以下内容写成文档并存档：候选因子清单、验证用的数据窗口、判定门槛、统计口径、以及"如果不可判定怎么办"
验证时严格按存档执行，禁止事后调整任何规则
规则改动 = 实验 identity 改变 = 必须整个重跑，旧结论作废

预注册的本质不是仪式感，是把研究者的自由度（researcher degrees of freedom）在结果出来之前锁死。你在第二讲见过的"参数可调的叠加因子"是在因子内部消耗自由度；"看完结果再决定用什么基准、什么窗口、什么门槛"是在验证环节消耗自由度——后者更隐蔽，杀伤力更大。

3. 验证管线的四个构件

以本课程配套的真实管线为例（AIResearchPlatform，与工厂分仓、独立拉数）：

构件一：独立数据复算

验证面板不从战役面板拷贝，而是从数据源独立拉取重算。原因直接来自实战：carry 因子 +18% 的双重复权 bug 藏在战役面板的构建代码里——如果验证直接消费这份面板，bug 会被原样继承，纠错永远不会发生。（口径问题的专题在第三章。）

构件二：预注册门槛

真实战役用到的门槛示例（写入预注册文档后才允许接触 OOS 数据）：

DSR（Deflated Sharpe Ratio）：对 Sharpe 做多重检验校正——你试了 N 个策略，显著性门槛就要相应抬高。N 越大，门槛越高，头名越难过线
主副裁判：主判据之外设一个口径不同的副判据（如价格口径 vs 资金费率口径），两者不一致时不得判"支持"
分年度约束：不允许"全部收益来自某一年"的结构——单年贡献超过上限即否决，防止把一段行情当成了规律
正交化对手：候选必须对已知风格因子（动量、价值、低波、流动性）做正交后仍有残差 alpha；被已有因子解释掉的部分不算新发现

构件三：样本外一次性判定（OOS 一次性）

样本外数据是消耗品：每用一次就被污染一次。规范是一段 OOS 窗口只允许做一次判定——

判 SUPPORTED：进入下一阶段（影子监控，第四章）
判 REJECTED：结论存档，该假设族进入"禁止复测"清单（防止换个马甲再试一次 OOS）
判 INCONCLUSIVE（如功效不足）：冻结，只能等新的前向数据或新预注册，不允许降低门槛重试

构件四：留痕与可复现

验证报告落盘存档（数据指纹、代码版本、门槛数值），任何人任何时刻可复算。这不仅是科学规范，也是给自己未来的保险——第九讲会讲到，多个重大勘误（包括 +18% 假 carry）正是因为"历史结论可以复算"才被发现和纠正的。

4. 一次性判定的三份真实判决书

这套制度产出的三种结局，每种都有真实案例，构成接下来两讲的内容：

判决	案例	数字	讲次
REJECTED	业绩预告事件窗	OOS 价差 −0.49%（方向反），发现期亦无效	第六讲
INCONCLUSIVE	指数调样流	OOS +1.00%，p=0.062——功效不足，冻结	第六讲
SUPPORTED	crypto funding carry	OOS +2.37%/年，p<0.0001，八项子检验全过	第四章

注意三个数字的相对量级：唯一活下来的策略，年化 2.37%；而被否决的两个，发现期的数字都远比它耀眼。真 alpha 又小又难看，假 alpha 又大又合理——这不是巧合，是这个行业的结构特征，后面每一讲都会回到这一点。

5. 本讲小结

工厂回测是描述，独立验证是判决；差别在制度不在算力
多重检验是循环挖掘的主导性失效模式，预注册是唯一的制度化解药
管线四构件：独立数据、预注册门槛（DSR/主副裁判/分年度/正交化对手）、OOS 一次性判定、留痕可复现
三种判决各有真实案例：REJECTED（−0.49%）、INCONCLUSIVE（+1.00%, p=0.062）、SUPPORTED（+2.37%/年）

思考题：你的同事说"DSR 门槛太严了，我们 290 个因子里一个都过不了，不如放宽到 10% 显著性水平"。请指出这段话里两处逻辑错误。（提示：门槛的功能是控制什么？改门槛等于做了什么？）

RD-AGENT 第三讲 · LLM 写的因子长什么样：AI 的研究品味

Henry Lin — Wed, 14 Oct 2026 02:33:12 +0000

第三讲 · LLM 写的因子长什么样：AI 的研究品味

第一章第三讲（本章收尾）
学习目标：通过真实案例建立对 LLM 因子假设质量的判断力；识别"伪逻辑假设"的特征；理解代码质量与逻辑质量的不对称。

1. 开场

前两讲我们隔着玻璃看工厂：循环怎么转（第一讲）、成绩单怎么读（第二讲）。这一讲开门进去，把 AI 写的因子摆上解剖台。材料全部来自真实战役产出（A 股 290 因子池、期货两轮、crypto 两轮）。

先给本章的总结论，然后逐类验证它：

AI 的代码比它的金融直觉诚实得多。 代码 bug 是显性的——跑不通就报错，反馈即时；逻辑空洞是隐性的——假设再空，回测照样吐出一个数字，而这个数字还会进入排行榜参与选拔。

2. 案例一：像样的——但要看清"像样"的来源

期货面板上真正有产业逻辑的因子，集中在人给的两列特有数据上：基差（期现价差） 和 持仓量。例如基于期限结构的假设：深度贴水（远月大幅低于近月）通常反映交割压力或悲观预期，构造基差 z-score 做均值回复或方向暴露。

这类因子值得肯定——但请注意它的产出方式：数据边界是人画的。人把 $basis、$oi 放进面板，AI 才有素材往这个方向想；没有这些列，它的假设空间就退化为纯量价。这个现象在三轮战役里反复出现：AI 提出的"最有金融味道"的假设，几乎总是在人准备好的特色数据附近打转。

结论一：AI 的研究品味上限，由数据面板的质量决定。它做的是"在给定素材空间内的高效搜索"，不是"开辟新的素材空间"。

2.1 一个必须讲的反面注脚：carry 因子的 +18% 是假的

基差/期限结构方向上，工厂曾产出一个发现期年化 +18% 的 carry 策略——是全部战役中最亮眼的数字。后来审计发现这是双重复权 bug：面板构建时收益在换月处被扣减了两次。纠错后重算：+3.18%，t=0.66，不显著。

一个数字从"接近可部署"到"统计噪声"，中间只隔一次口径审计。这个故事第九讲整讲展开（期货面板五坑），这里先立一块碑：

在数据口径被验证之前，最像真的数字往往是最危险的数字。 +18% 之所以差点骗过所有人，恰恰因为它"看起来太合理了"——期限结构carry 本来就该有收益，bug 放大的数字落在了预期之内。

3. 案例二：平庸的——伪逻辑的识别

排行榜中段大量存在这一类。以 turn_of_month_rank（换月效应）和 overlay_reversal_liquidity（反转×流动性叠加）为代表——自报超额 +26.5% 和 +14.9%。

伪逻辑假设的三个特征，用反转×流动性叠加举例：

自由度过高：两个已知现象（短期反转、流动性溢价）相乘或取交集，参数（回看窗口、过滤阈值、分位边界）全是可调的。参数空间越大，回测越容易找到好看的一格。
故事讲得好但机制模糊："流动性差的票反转更强，因为套利资金进不去"——听起来合理，但它同时解释得了相反的结果（"流动性差的票动量更强，因为信息扩散慢"）。能解释任何结果的机制，等于没有机制。
评估次数多：turn_of_month_rank 15 次评估、14 个版本——循环反复碰它，本身就是"数字靠多次尝试磨出来的"的信号（第二讲的 combos/versions 原则）。

对照一个正例：crypto 两轮战役后经独立验证确认的真信号是 vol_20d（波动率）和 reversal_5d（5 日反转）——注意，这两个是文献里被反复验证过几十年的现象。AI 循环绕了一大圈，最后活下来的还是教科书因子。这不是巧合：教科书因子经过了几十年、几十个市场、无数研究者的"多重检验"，而循环内新造的叠加因子只经过了排行榜这一次选拔。

结论二：对 LLM 生成的新颖组合因子，默认假设是"参数搜索的产物"；举证责任在因子一方。经受过跨市场、跨年代检验的经典现象，先验天然高一档。

4. 案例三：翻车的——以及翻车之后发生了什么

循环里的失败分两种，都有教学价值。

跑不通型：代码报错（对齐错误、缺列、边界溢出）。这一类 AI 处理得很好：读报错 → 定位 → 自动修复 → 重跑。工程上无可挑剔。

隐性翻车型（危险得多）：代码跑通、回测数字完美，但逻辑上用了未来数据（前视偏差）或口径错位。这一类工厂自身没有任何警报机制——它和好因子在排行榜上长得一模一样，只能靠独立验证暴露。期货第一轮 136 个试验在预注册验证下全灭，其中一部分就属于这类：工厂口径的数字看不出问题，独立管线一复算就原形毕露。

于是有本讲最重要的一条不对称性：

代码错误：显性、即时反馈、AI 能自修。
逻辑错误：隐性、无反馈、AI 毫无察觉、人若不验也不知道。
循环加速了第一种错误的消亡，同时加速了第二种错误的繁殖。

5. 本章总结

三章合起来，给"AI 因子工厂"一个诚实的画像：

维度	评价
代码实现	强。质量高、可自修、产出速度人无法比拟
假设生成	中。在给定数据素材内高效搜索，经典现象的再发现可靠，新颖假设多为高自由度重组合
数据口径	弱。双重复权、基准伪影这类问题 AI 完全无感
结果可信度	无内置机制。排行榜是多重检验发生器，头名不可信
真正的瓶颈	不是"AI 会不会想因子"，而是"人会不会验证"

290 个因子、自报超额最高 +26.5%、独立验证零存活——这就是第一章的终点，也是第二章的起点。下一章讲验证管线本身：预注册、DSR 门槛、主副裁判，看"全军覆没"这个结论是怎么被诚实且可复现地得出来的。

思考题：如果让你设计一个机制，让循环在生成阶段就自动过滤掉高自由度的伪逻辑假设，你会检查哪些特征？（提示：可调参数个数、假设能否被相反结果证伪、是否引用了特色数据列）

RD-AGENT 第二讲 · 看懂排行榜：IC、超额收益与因子版本

Henry Lin — Fri, 09 Oct 2026 02:10:20 +0000

第二讲 · 看懂排行榜：IC、超额收益与因子版本

第一章第二讲
学习目标：掌握排行榜四个核心指标的含义与陷阱；理解因子"组合/版本"结构；认清"自报成绩"的三个口径问题。
本讲所有数字均为工厂自报口径（工厂内置回测，未经独立验证）——这本身就是本讲的主题之一。

1. 一张真实的排行榜

A 股战役的因子排行榜，共 290 个因子。按自报超额收益排序的头部：

因子	类别	自报超额	评估次数(combos)	代码版本(versions)
asset_growth_neg	其他	+26.5%	4	2
low_beta_rank	其他	+26.5%	4	2
turn_of_month_rank	流动性	+26.5%	15	14
low_volatility_rank	波动率	+20.8%	2	2
overlay_reversal_liquidity	动量/反转	+14.9%	2	1
ep_rank	价值	+14.7%	22	18
momentum_20d / reversal_5d	动量/反转	+14.7%	23	33

第一讲说过：这些数字一个都不能直接信。但要拆穿它，得先能读懂它。

2. 四个指标：直觉与陷阱

IC（信息系数）

因子值与未来收益的截面相关系数，逐日计算。

直觉：预测方向准不准。IC=0.03 在 A 股已经是值得认真对待的水平。
陷阱：平均 IC 可能被少数极端交易日撑起来——50 个交易日里 49 天接近 0、1 天 0.3，平均出来 0.006 之外还会给你一个漂亮的最大值。

Rank IC

用秩相关（Spearman）替代 Pearson。

为什么量化界更爱它：对离群值稳健。一只票涨停 20% 和涨 10%，对秩来说都是"第一名"。
实务习惯：看 Rank IC 的符号一致率（正的天数占比），比看均值更早暴露问题。

ICIR

IC 的均值除以 IC 的标准差——预测的稳定性。

直觉：IC 均值 0.05、ICIR 0.5 的因子，好于 IC 均值 0.08、ICIR 0.2 的。
陷阱：ICIR 对换手率只字未提。一个每天全换仓的因子 ICIR 可以很好，但手续费吃光一切。

超额收益（excess return）

相对基准的组合收益之差。直觉最直白，陷阱最多：

基准是谁？ 同一因子换个基准，超额可以换符号。本战役期货用 23 品种等权指数；A 股战役曾用沪深 300——后来发现基准里剔除了涨停股，导致"超额"有相当部分是伪影（第八讲详述）。
成本算了吗？ 工厂回测默认口径未必含真实手续费。
怎么多空？ qlib 0.9.6 的回测器三层封死做空——工厂的多空数字实际来自一套向量化旁路计算，口径又换了一次。

3. 因子的"组合"与"版本"：为什么一个因子有十几条记录

排行榜上 momentum_20d 有 23 次评估、33 个代码版本。三层结构：

因子名：一个假设（"20 日动量"）
组合（combos）：这个假设在不同时间窗/不同循环里被反复实现和评估的记录。23 次评估意味着循环在 23 个不同上下文里碰过它
版本（versions）：代码迭代历史——AI 修过 bug、调过参数、换过实现方式

这里有一个排行榜不会明说、但本课程反复使用的概念：多重检验。

循环碰了它 23 次，你看到的是 23 次里最好的一次。抛 23 次硬币，出现一次连正不奇怪；测 23 次因子，出现一次好看的超额同样不奇怪。290 个因子、每个平均若干次评估——这张排行榜是一台多重检验发生器。

第二讲思考题的答案现在可以说了：循环跑 1000 轮，排行榜第一名会更不可信——选拔基数越大，头名越可能是运气最好的噪声。这不是 AI 的问题，是人类研究员手动试 1000 个因子时同样会发生的问题；AI 只是把它加速了几十倍。

4. 自报成绩的三个"没告诉你"

基准是谁——超额是相对量，基准的每个口径决定（剔除规则、加权方式、可交易性假设）都会渗入超额。A 股战役的教训：基准剔涨停 → 做多侧天然避开涨停股 → "超额"里有一块是伪影。
成本算了吗——排行榜数字多为毛收益或半净口径。高换手因子（33 个版本的动量类是重灾区）净费后经常翻负。
测了几次——如上节。数字好看的头名，先查它的 combos/versions 数：评估次数越多的因子，其最佳记录的运气成分上限越高。

一个对照实验预告：这 290 个因子的排行榜头名，后来在独立验证管线里按预注册规则复测——没有一个活下来（第五讲）。不是因子全军覆没，而是"排行榜排名"和"真有效性"之间的相关，远比直觉弱。

5. 本讲小结

IC 看方向、Rank IC 抗离群、ICIR 看稳定，三者都不含成本信息
combos/versions 列是"被测了几次"的直接证据，读排行榜先读这两列
排行榜是多重检验发生器：290 个因子的头名自报 +26.5%，独立验证后全军覆没
三个必问：基准是谁、成本算了吗、测了几次

思考题：如果你只能往排行榜加一列来防止自己被骗，加什么列？（本课程的选择：预注册判定结果——先想清楚再对照答案）

RD-AGENT 第一讲 · AI 因子工厂是怎么运转的

Henry Lin — Tue, 06 Oct 2026 01:08:14 +0000

第一讲 · AI 因子工厂是怎么运转的

《AI 挖因子时，如何不被自己和 AI 骗到》第一章第一讲
学习目标：理解 RD-Agent 类因子挖掘循环的五个步骤、人与 AI 的分工边界；能看懂一次循环的执行记录。

1. 开场：一台真实跑过的机器

这门课的素材全部来自一次真实战役：2026 年 9 月，在一个 23 个品种的期货面板上（2022-12 至 2026-09，约 918 个交易日），用 RD-Agent 循环自动挖掘因子。第一轮（13 品种）产出 136 个试验；第二轮到第一个窗口收口时跑了 40 个循环，产出 200 多个工作区。

先给结论，这门课后面每一讲都在论证它：

AI 因子工厂替代的是"手"，不是"判断"。 它一晚上能写几十个因子的代码并跑完回测，这是真的；但提出的假设是否成立、结果能不能信，每个环节的判断仍然是人的——而"结果能不能信"恰恰是最容易做错的一环。

2. 循环的五步

构思(gen) → 写码(coding) → 回测(running) → 反馈(feedback) → 记录(record)
     ↑____________________ 记忆在循环间传递 ____________________|

第一步：构思（direct_exp_gen）。 LLM 读取历史因子的反馈记录（"上一个动量因子在 2024 年段失效"），提出新的因子假设：一段动机陈述、公式草案、预期方向。这一步的输入还包括人定义好的数据面板——23 个品种 × 21 列（开高低收、成交量、结算价、基差、持仓量等）。

第二步：写码（coding）。 把假设翻译成可执行的 pandas 代码，在面板上计算因子并落盘。这一步是最"工程"的，也是 LLM 表现最稳的一步——代码要么跑通要么报错，反馈是即时的。

第三步：回测（running）。 因子被送进 qlib 回测，产出超额收益、IC 序列等指标。注意：回测口径（基准是谁、能不能做空、成本怎么算）是人预先写死在配置里的，AI 改不了。本战役的基准是 23 品种等权指数（EQWI-23），多空评估走向量化口径。

第四步：反馈（feedback）。 LLM 读回测结果，写"实验笔记"：为什么有效/失效、下一步试什么。这是循环里最有意思的一步——AI 给自己写复盘。

第五步：记录（record）。 归档到工作区，供下一轮构思读取。

关键机制：记忆。 第 N 轮的构思会读第 N-1 轮的反馈，所以它不是独立随机撒点，是有累积的。但注意：记了笔记 ≠ 吸取了教训——第三讲我们会看到，它的"演化"很大程度是在已知量价现象上做重组合。

3. 一次循环执行记录长什么样

打开 Console 的运行记录页，选期货第二轮首窗那条 40 循环的会话，展开任意一个循环，五个步骤目录依次排开。两个值得驻足的位置：

构思步的输出：一段完整的因子假设文档——"基于期限结构的假设：基差深度贴水时……预期空头方向收益……"。念出来你会发现：像一份入门研究员的开题报告。
反馈步的输出：读回测数字后写的自我总结——"该因子在加入波动率过滤后失效，推测原因为……，建议下一轮尝试……"。

快速往后翻：Loop_15、Loop_30 的构思内容明显在回应前面的反馈。这就是记忆在起作用。

4. 工厂的日常：一半时间在跟基础设施搏斗

真实的执行记录里还有大量"不体面"的部分。第二个窗口刚启动就撞上 LLM 服务端故障：HTTP 200 但返回空响应，持续断续 24 小时以上，一个窗口约一半产能被吃掉。应对是工程化的：断点续跑监督器、30 次重试加退避、会话恢复。

这一节想传达的不是具体技术，而是一个预期管理：

宣传视频里的 AI 工厂是永动印钞机；执行记录里的 AI 工厂，日常一半时间在报错重试。评估这套系统的正确单位不是"它多聪明"，而是"扣除故障后，单位成本产出多少个可检验的假设"。

5. 人站在哪里

决策	归属	本战役的实际选择
数据面板（品种、列、复权方式）	人	23 品种、结算价衔接复权、含基差/持仓量
回测口径（基准、成本、能否做空）	人	EQWI-23 等权基准、向量化多空
时间预算、何时开停	人	6 小时窗口、监督器收口
因子假设本身	AI	40 循环 × 每循环若干因子
代码实现	AI	自动生成、自动修错
信不信结果、怎么验证	人	独立验证管线（第二章主角）

注意最后一行：这是整张表里最难、也最容易被略过的一格。工厂自带的回测会告诉你每个因子的超额收益——290 个因子的排行榜，最高的自报超额 26%。这些数字一个都不能直接信。 为什么、怎么验，是第二章的内容。

6. 本讲小结

五步循环：构思→写码→回测→反馈→记录，记忆跨循环传递
AI 的强项在写码（反馈即时、对错分明），弱项在假设的金融逻辑（第三讲展开）
人负责数据、口径、预算和验证；AI 负责假设和实现
290 个因子、自报超额最高 26%——把这个数字记住，第二讲拆它

思考题：如果让循环跑 1000 轮而不是 40 轮，排行榜第一名的自报超额会更可信还是更不可信？为什么？（答案方向在第二讲）

量化开源项目v6.0已经更新

Henry Lin — Mon, 05 Oct 2026 02:45:12 +0000

henrylin99 / quantitative_analysis

量化分析

多因子选股系统

一个面向 A 股市场的量化分析系统，涵盖实时行情分析、因子计算、机器学习建模、交易信号、组合管理、风险管理、回测验证等完整链路。v4.0 采用 React 前端 + Flask API 架构，数据层延续 Parquet + SQLite，零外部中间件依赖（无需 MySQL / Redis），克隆即可运行。

项目定位

本项目是用于学习和二次开发的多因子选股系统原型，适合继续补齐后再扩展使用。

定位：多因子选股系统原型，适合继续补齐后再扩展使用
适用场景：量化入门学习、策略研究、二次开发
不适用场景：直接用于实盘交易

能力现状（已实现 / 部分实现 / 未实现/未开放）

模块	状态	说明
日线/分钟线数据中心	已实现	任务提交、状态轮询、历史展示（Parquet 存储）
AI 智能工作台	已实现	自然语言查数据/更新数据/建宽表/算因子/按区间批量构建因子数据/因子组合回测/训练模型，流式对话，SQL 只读约束
因子计算与表达式	已实现	内置因子 + Alpha191 因子库（191 个）+ 自定义表达式（白名单校验），point-in-time 公告日打点
ML 建模与预测	已实现	XGBoost / LightGBM / RandomForest，防泄漏训练管线
基础选股评分	已实现	基础因子评分与 ML 集成评分
投资组合管理	已实现	投资组合页面已具备真实创建、详情、持仓增删改、组合删除和优化结果落库能力
回测验证	已实现（研究级）	t+1 成交、涨跌停/停牌约束、逐日 mark-to-market 净值已实现；卖出印花税计入成本；注意股票池完整性依赖重新下载含退市股的 stock_basic
报告中心	部分实现	报告列表与生成功能入口已接通，导出与订阅派发仍在补齐
实时行情分析	部分实现	React 前端六页已可用（指标/信号/监控/风险/报告/推送）；分钟数据依赖本地同步任务
试用数据面板	已实现	市场简报、财务健康度、资金流统计、个股雷达/全景、板块热力图（`/trial/*`）
Text2SQL 查询	部分实现	只读查询可用（统一只读校验），模板生态待完善
实盘交易对接	未实现/未开放	无券商接口，不构成投资建议

功能概览

🤖 AI 智能工作台

通过自然语言对话完成查询分析与系统操作，页面入口：导航栏「AI 工作台」（/ai-workbench）。

数据查询分析：大模型生成只读 SQL 查询本地宽表数据（最新交易日快照），自动查看表结构、修正 SQL
代办系统功能：下载/更新数据（需在 .env 配置 TUSHARE_TOKEN）、构建大宽表、计算因子、按日期区间批量构建因子数据（含 Alpha191 全套，后台任务）、基于因子的组合回测与结果查询、创建自定义因子、训练与预测 ML 模型、查询任务进度
对话示例：「构建 alpha191 从 2026 年 1 月至今的数据」「以沪深300为基准，基于 alpha_001 因子回测 2026 年 9 月」——长耗时操作自动转后台任务并返回任务号，追问进度即可
安全约束：SQL 仅允许单条只读 SELECT（复用 text2sql 校验与独立只读查询库）；动作类工具可整体切换为只读模式；大宽表构建保留 18:00 校验
配置方式：在 .env 中设置 LLM_API_KEY / LLM_BASE_URL / LLM_MODEL 即可，兼容 DeepSeek、通义、GLM、Kimi、OpenAI 及本地…

View on GitHub

量化开源软件v3.0发布

Henry Lin — Sun, 23 Aug 2026 04:40:49 +0000

下载地址：
https://github.com/henrylin99/quantitative_analysis/releases
视频地址：
https://youtu.be/p0iJxGveW60

v3.0.0 — AI 智能工作台
核心变更
新增大模型驱动的「AI 智能工作台」（/ai-workbench，导航栏入口）：用自然语言对话完成行情数据查询分析，并直接代办数据下载更新、大宽表构建、因子计算、机器学习模型训练与预测等系统已有功能。

从此，"更新日线数据"、"合并大宽表"、"最新交易日涨幅前10" 这类需求，一句话即可完成。

主要特性
🤖 自然语言查数据：大模型自动查看表结构与样例数据、生成只读 SQL 查询本地宽表快照（行情、估值、资金流、技术指标），查询失败自动修正重试，多股对比用表格呈现
⚙️ 代办系统功能：单个/批量数据任务下载更新（自动推导增量窗口，防止留缺口）、大宽表构建（保留 18:00 校验）、因子计算与自定义因子创建（公式白名单校验）、ML 模型训练/预测/进度轮询
🏦 财务三表 vip 接口：income_vip / balancesheet_vip / cashflow_vip 按报告期增量下载，替代原逐只循环——三张表从上万次接口调用降到个位数，并自动覆盖财报更正
💬 流式对话体验：SSE 打字机式输出；每次工具调用全程可视化（工具名、参数、结果、耗时）；会话持久化，刷新/重启后可回放历史；支持「操作模式 / 只读模式」一键切换
🔐 安全约束：SQL 仅允许单条只读 SELECT（复用只读校验链 + 独立隔离查询库，物理上无法写入数据）；只读模式下动作类工具整体禁用；危险任务不允许通过 AI 触发；full_refresh 全量刷新需用户确认
🔌 一行配置接入大模型：兼容 DeepSeek、通义、GLM、Kimi、OpenAI 以及本地 Ollama 的 OpenAI 兼容接口，.env 三个变量即可切换

DeepSeek Harness (`dsh`) Agent 开发教程

Henry Lin — Thu, 20 Aug 2026 11:02:14 +0000

DeepSeek Harness (`dsh`) Agent 开发教程

本教程面向想开发 Agent 的开发者。这里的"开发 Agent"有不同含义，dsh 对应不同做法：组合一个可运行的 agent（声明式）、写程序驱动 agent 的协议层/UI、"造"一个自己的 agent 驱动，以及让 agent 能委派子代理。教程把每条路都走一遍，并在开头讲清楚该选哪条。

Agent 在架构中的位置
五个核心概念
三条开发路径怎么选
路径一：组合一个可运行 Agent
路径二：程序化创建与驱动 Agent
路径三：实现自定义 Agent 驱动
拦截与定制：不改 loop 的扩展点
委派：子代理（Subagent）
测试与验证
常见问题排查
完整示例：多角色量化研究 Agent

1. Agent 在架构中的位置

dsh 是插件驱动的 harness。Agent 不是内置的"黑盒"，而是由一组插件搭出来的 spine。每个部分都可以替换，包括 agent 主循环本身。

由六个包构成的核心 spine（各自职责见 docs/subsystems/core.md）：

包	职责	ctx key
`core/session`	追加式会话事件日志，事实源	`ctx.sessions`
`core/system-prompt`	prompt 段 + 工具 schema 组装	`ctx.systemPrompt`
`core/tools`	工具注册表 + 执行管线	`ctx.tools`
`core/agent`	`Agent` 接口、活注册表、`agent/*` 事件、initiator	`ctx.agents`
`core/agent-loop`	具体驱动（唯一实现）	`ctx.agentLoop`
`core/scope`	每个 agent 的独立注册空间	`agent.ctx`

重要约定：扩展插件依赖 agent 包，从不对 agent-loop 编程，所以 loop 保持可替换。默认的完整 spine 打包在 agent-spine-demo 里，任何 app 包通过它组合出一个能跑的 agent。

一次 turn 的流程（完整时序 docs/agent-lifecycle.md）：

followup(wake) → running → turn/start → claim inbox → agent/pre-step (waterfall)
  → step/start → user/message → agent/request (waterfall) → llm/stream
  → assistant/chunk* → assistant/message → tool/call* → tools/execute → tool/result*
  → step/end → turn/end → idle

turn/*、step/*、user/message、assistant/*、tool/* 是 durable 会话事件，追加进日志、重启可重建；agent/*、tools/* 是 live 运行期事件，负责状态、拦截、排队。

2. 五个核心概念

2.1 `Agent`：活 agent 的编程面

Agent 是每个插件（UI、钩子、编排器）对着写程序的接口（以下为方法清单节选，完整声明在 packages/core/agent/src/types.ts）：

interface Agent {
  readonly id: SessionId        // 与会话共享的唯一身份
  readonly options: AgentOptions // provider / model / maxTokens
  readonly session: Session      // 活会话；日志是 durable 事实源
  readonly inbox: Inbox          // 待处理消息的持久投影
  readonly status: AgentStatus   // 'idle' | 'running'
  readonly ctx: Context          // 这个 agent 私有的 scoped context

  send(message: UserMessage, target: InboxTarget, wakeup: boolean): void
  followup(message: UserMessage): void   // 排队下一 turn 并唤醒
  steer(message: UserMessage): void      // 驱动最近一个 step
  inject(message: UserMessage): void     // 加入模型上下文，不唤醒
  cancel(cause: AgentCancelCause, options?: CancelOptions): void
  whenIdle(): Promise<void>
  runMaintenance<T>(task: (signal: AbortSignal) => Promise<T>): Promise<T>
}

2.2 `ctx.agents`：注册表与创建入口

ctx.agents 追踪所有活 agent，create() / resume() 创建时调用已注册的 factory：

const handle = await ctx.agents.create({
  sessionId: SessionId('my-session'),
  agentOptions: { provider: 'deepseek-official', model: 'deepseek-v4-flash' },
})
// 之后：handle.agent.followup(...)；结束时 await handle.dispose()

返回的 AgentHandle 是唯一拿得到 dispose 能力的地方。ctx.agents.get(id) 只返回裸 Agent，没有归属就不允许拆掉它。

2.3 三种送达方式：followup / steer / inject

方法	目标	唤醒?
`followup()`	`next-turn` inbox，作为自己 turn 的唯一普通消息	是
`steer()`	`next-step` inbox，运行中 agent 在下一个 step 醒来即消费	是（idle 时开一个 turn）
`inject()`	`next-step` inbox，注入的上下文	否（idle agent 保持 idle）

followup() 不返回任何会关联到未来结果的句柄；MessageId 只标识 durable 的在册插入/认领/丢弃事实。VM 要确认工作"跑完了"，用 whenIdle()（agent 整体静止），而不是用一个消息的 turn 结束去猜。

2.4 `SessionId` 与品牌化 ID

跨包的 id 都是 branded 类型：SessionId、CallId、JobId 结构上都是 string，但类型上不通用。用各自工厂构造，比如 SessionId('abc')。别用裸字符串在 ctx.agents 里查。

2.5 Initiator（发起人）作用域

ctx.agents.withInitiator(agent, op) 给一段异步调用链挂上"发起人是这个 agent"的因果归属；currentInitiator() 读取它。适用于日志、遥测、归属。注意：在场不等于授权，不做 liveness 证明。

3. 三条开发路径怎么选

你的目标	路径	典型产出
做一个能跑、行为可调的产品 agent	组合（拼 cordis.yml + spine + leaf）	一条 `cordis.yml`，一个 app bin
写一个"壳"来操作 agent（CLI/UI/协议/编排）	程序化驱动	ACP / JSON-RPC 桥、UI 插件、编排器
换掉默认的 turn 驱动逻辑	自定义驱动	实现了 `Agent` 契约的服务
让 agent 内部能互相委派	子代理（第 8 章）	新增 `ctx.subagents` provider
只改某个环节行为（拦截/注入/钩子）	第 7 章扩展点	监听器插件，不动 loop

按顺序：先组合，再驱动，最后才是改 loop。绝大多数需求落在前三条，自定义 loop 是极少数。

4. 路径一：组合一个可运行 Agent

4.1 原理：bundle 组合 + leaf 提供 backend

dsh-agent-spine-demo 把一切 agent 通用件打包成一个 bundle：

装好：llm 抽象、session、system-prompt、tools、subagent 注册表、agent-loop（具体驱动）、skill 本地 provider、AGENTS.md 加载器、todo/jobs 工具…
故意不装：LLM 适配器、bash 执行器、非本地 skill provider、app 入口。这些由加载它的 leaf 补上 —— 这正是"seam"的作用：缝合处（spine）与可替换件（backend）分开。

参考真实 leaf：examples/headless-agent/cordis.yml —— 一次性编码 agent 的完整组合。

4.2 用 `agents` 配置项预先创建 agent

spine 把 agents 列表转发给 dsh-agent-loop，启动时就在注册表里建好 agent：

- id: agent-spine
  name: '@deepseek-ai/dsh-agent-spine-demo'
  config:
    agents:
      - id: main
        provider: deepseek-official
        model: deepseek-v4-flash
        cwd: !!js process.cwd()
    workspaceContext:
      maxBytes: 65536
    persona: |
      You are headless-agent, a coding assistant powered by the {{model}} model.
      Verify your work by running the code or tests. Keep answers brief.

像 ACP 这类 app 反而不预建 agent：session/new 时动态 create()（见第 5 章）。

4.3 补 leaf：适配器、沙箱、持久化

组合里还要补三类 backend：

# LLM 适配器（真实 provider）
- id: llm-deepseek
  name: '@deepseek-ai/dsh-llm-deepseek'
  config:
    models:
      - id: deepseek-v4-flash

# bash 执行链：子进程 + bash 后端
- id: subprocess
  name: '@deepseek-ai/dsh-subprocess-local'
- id: bash
  name: '@deepseek-ai/dsh-bash-local'
  config:
    timeoutMs: 60000

# 会话持久化 JSONL
- id: persistence
  name: '@deepseek-ai/dsh-session-persistence-jsonl'
  config:
    root: './.sessions'

工具链、审批策略、沙箱、web/fetch/lsp 等按需加；dsh-web-app / dsh-headless 这类 bundle 已经把常见组合打包好了。

4.4 运行与验证

# 从源码跑源码里的组合
pnpm dsh --profile headless "把仓库的 packages/ 结构总结成树"

# 自己的组合：用 expose 一个 bin 的 app 包（如 dsh-jsonrpc-agent）加载它
pnpm dsh-jsonrpc-agent path/to/cordis.yml

# 只检查组合出来的树，不启动
pnpm dsh --profile headless --dump-config

组合能跑之后，agent 就"是"这一棵插件树。想要不同能力就改组合，而不是改代码。

5. 路径二：程序化创建与驱动 Agent

这条路径是 UI、协议桥、编排器的写法：进程里已经有一个 spine，插件在 apply() 里用 ctx.agents 操作活 agent。

5.1 最小协议桥骨架

import type { Context } from '@deepseek-ai/cordis'
import { createUserMessage } from '@deepseek-ai/dsh-llm'
import { SessionId } from '@deepseek-ai/dsh-session'

export const name = 'my-protocol-bridge'
export const inject = ['agents', 'sessions', 'sessionPersistence']

export function apply(ctx: Context) {
  // 外部协议 -> agent
  async function onPrompt(text: string) {
    const handle = await ctx.agents.create({
      sessionId: SessionId(`client-${nextId()}`),
      agentOptions: { provider: 'deepseek-official', model: 'deepseek-v4-flash' },
    })
    handle.agent.followup(createUserMessage({
      content: [{ type: 'text', text }],
      source: { kind: 'user' },
    }))
    return handle
  }

  // agent -> 外部协议：消费 durable 会话事件
  ctx.on('session/event', (_session, event) => {
    if (event.type === 'assistant/chunk' && event.data.chunk.type === 'text-delta') {
      sendToClient(event.data.chunk.text)
    }
  })
}

要点：

只对 session/event 的 durable 流做回放式消费，不用 agent/* 的 live 事件重建会话 —— 那会丢掉可重建性。
低层协议请求拿到的是入队回执，不是结果句柄：followup() 不返回结果。要等"跑完了"由调用方显式持有 receipt→whenIdle 区间，别用 MessageId 去关联 turn/end。
要优雅停机：handle.dispose() 会 stop + await 退出，到达 quiescence。参考实现：dsh-acp。

5.2 resume：加载持久化会话

const handle = await ctx.agents.resume({
  resumeSessionId: SessionId('some-existing'),
  agentOptions: { provider: 'deepseek-official', model: 'deepseek-v4-flash' },
})

这会先加载持久化会话再在其上恢复 agent —— fork、断点续跑、历史会话回放都走这里（sessions.create(id, { seed }) 可做重放）。

5.3 前端 / UI 插件

UI 从 session/event 渲染，把用户输入用 followup() / steer() 送去。一个最小 UI 插件就是：监听 assistant/chunk 画 token 流，用户敲字 → followup()。完整示例与 ConversationNodeDefinition 注册（给内置 Web Client 加业务节点）见添加一个 Chat 节点与 extension-cookbook 的 UI 段。

6. 路径三：实现自定义 Agent 驱动

默认 loop（dsh-agent-loop）是 Agent 契约的唯一实现，但契约本身可以再实现。为什么极少需要：绝大多数"不同行为"都能落到第 7 章的扩展点上。只有当你要重写 turn 驱动语义（认领、重试、步进策略）时，才需要自己的驱动。

实现契约的方式：构造自己的服务，实现 AgentFactory（create / resume），通过 ctx.agents.setFactory(factory) 注册，会让 ctx.agents.create/resume 走你的实现而不是默认 loop。每个 agent 的注册与生命周期要走 ctx.agents.withInitiator()、agent/* 事件词汇和 session/event 日志纪律 —— Agent 接口就是你和所有消费者之间的完整契约，接口本身在 packages/core/agent/src/types.ts。

先确认你知道自己在放弃什么：换驱动等于退出产品配好的编排（预设、checkpoint、计费、重放），这些通常比"自定义步进"更有价值。

7. 拦截与定制：不改 loop 的扩展点

新行为 = 往文档化扩展点挂监听器。没有任何一行要改 loop（他们的微内核承诺）。

7.1 `agent/*` 事件速查

事件	规格	用途
`agent/created`	emit	agent 发布完成（scope 过滤）
`agent/disposed`	emit	agent 离开注册表
`agent/error`	emit	step/turn 出错
`agent/inbox/inserted` / `claimed` / `discarded`	emit	inbox 变更通知
`agent/pre-step`	waterfall	否决 step 或替换进入的 messages
`agent/request`	waterfall	替换请求配置（`LlmCallConfig`）
`agent/request-error`	waterfall	失败恢复，返回 `retry` 不调 `next()` 即认领
`agent/turn-stopping`	serial	turn 无延续时收尾；可 steer 下一步

waterfall 监听器必须调 next() 放行，忘了它就短路整条链。

7.2 钩子示例：拦请求换模型

import type { Context } from '@deepseek-ai/cordis'
import type { LlmCallConfig } from '@deepseek-ai/dsh-llm'

export const name = 'model-switcher'
export const inject = ['agents']

export function apply(ctx: Context) {
  ctx.on('agent/request', async (_payload, next): Promise<LlmCallConfig> => {
    const config = await next()          // 取机器本来要用的配置
    return {
      ...config,
      model: config.model?.startsWith('deepseek') ? config.model : 'deepseek-v4-flash',
    }
  })
}

7.3 注入模型上下文 vs 唤醒

要"塞给下一次请求"动态上下文（AGENTS.md 变更、文件改动通知、工具结果摘要），用 agent.inject()；它 appends durable 上下文，不唤醒 idle agent：

try {
  agent.inject(createUserMessage({
    content: [{ type: 'text', text: 'NOTE: AGENTS.md changed; re-read it.' }],
    source: { kind: 'plugin', plugin: 'my-watcher' },
  }))
} catch {
  // agent 已 dispose 时忽略（guarded）
}

7.4 每个 agent 一套能力：预设与 scope

ctx.agentPresets.mount(agentCtx, id) 把某一套插件组合（预设）装进这个 agent 的 scoped context；ctx.agentPresets.composeFrom(agentCtx, parentCtx) 让子 agent 复用父 agent 的同一代组合。注册表、prompt 段、工具都可以 per-agent 隔离，是"给不同会话不同工具集"的手段。预设与服务行 isolate realm 组合时，服务对组外不可见。

8. 委派：子代理（Subagent）

子代理是可选的能力缝，不是 loop 的一部分。特点：多个 provider 同时注册，按名字选（ctx.subagents），不同于 bash 的单执行器。Service Definition 在 dsh-subagent，Consumer 是 dsh-tool-subagent（把配置好的 provider 暴露给模型）。

8.1 注册一个 Provider

实现 SubagentProvider.start(request)（以及可选的 prepareContinuable）并注册进 ctx.subagents，用 provider 的 name 区分：

// provider 自身携带 name；注册返回 effect disposer，HMR 安全
const disposer = ctx.subagents.registerProvider(myProvider)

// 一次调用请求里最重要的字段（类型定义在 dsh-subagent）：
//   parent: Agent         生成者（派生 cwd/lineage/depth）
//   prompt: ContentBlock[] 送给子 agent 的 user message
//   signal: AbortSignal    启动前后统一取消通道
//   outputSchema / maxDepth / toolFilter / persona  需要匹配 capability flags

不支持的 start-time 能力（SubagentCapabilities）在 start 前就 fail loud，绝不 accept-then-ignore。

8.2 用 tool 把 provider 交给模型

headless 组合里典型一段：

# 两个 in-process provider：fresh spawn 与 fork
- id: subagent
  name: '@deepseek-ai/dsh-subagent'
- id: subagent-spawn-in-process
  name: '@deepseek-ai/dsh-subagent-spawn-in-process'
  config: { providerName: spawn }
- id: subagent-fork-in-process
  name: '@deepseek-ai/dsh-subagent-fork-in-process'
  config: { providerName: fork }

# 暴露为模型工具
- id: tool-subagent
  name: '@deepseek-ai/dsh-tool-subagent'
  config:
    provider: spawn
    toolName: subagent
    backgroundMode: continuable
    maxDepth: 1

模型通过 subagent / subagent_fork 等工具委托，子事件进入各自会话日志。continuable（可延续）子代理额外装全局 send_message 控制和子 scope 里的 report 返回通道——注意 fork 保持 one-shot，因为可延续子代理的 report/prompt 段先于 fork 复用的历史（见头文件注释）。

其他 shipped provider：-acp、-codex、-claude-code、-dsh-sdk。实现 provider 前先读 dsh-subagent README 的完整契约（features 发现、capability flags、start 校验顺序）。

9. 测试与验证

单元测试：pnpm run test（每文件 100% 是 CI 门槛）。先看 docs/testing.md 的分层约定。
真实组合测试：产品可见的 agent 行为，测试必须通过 Loader + 进程 boot 一个 cordis.yml，对手搭的 ctx.plugin(...) 单测不足够。agent-spine-demo 有个"library only" spine 可以插进这种测试。
快照回放（keyless）：pnpm run test:snapshot 重放记录过的会话，验证组装成产品的输出；非平凡的模型可见行为变化要在同一个 PR 里带上快照。
真实 API：pnpm run test:e2e（有 DEEPSEEK_API_KEY 才跑）。
操作手册级别验证：pnpm dsh --profile headless "任务" 把 agent 真的跑一遍。

自研驱动、协议桥、子代理 provider 这些"缝实现"，尤其要覆盖生命周期：创建、取消、dispose、HMR 卸载（注册必须可逆）、静默收敛。

10. 常见问题排查

现象	原因 / 处理
`create()` 拒绝 / 订阅不了 provider	没有 factory：leaf 没装 spine（`dsh-agent-loop`），或 `setFactory` 没被调用。先 `--dump-config`
dispatch 时 `UNKNOWN_MODEL` / `MISSING_CREDENTIAL`	组合里少了对应 provider 适配器，或凭证没配（参考 usage-tutorial 第 4 章）
新工具模型看不到	scope（preset/restrict）把它滤掉了；在 `agent.ctx` 里排查注册归属
忘了 `next()`	waterfall 直接短路：`pre-step`/`request` 类监听器必须放行
followup 后拿不到结果	`followup()` 本来就不返回结果；自己持有 receipt→`whenIdle` 区间才叫"跑完"
inject 后 agent 不动	`inject()` 不唤醒；要真的动起来用 `steer()`/`followup()`
子代理能力被拒	provider 的 `SubagentCapabilities` 没声明该能力（outputSchema/maxDepth/toolFilter/persona），start 前 fail loud
dispose 后还收到事件	自己持有了未释放的 effect/监听；注册要走 `ctx.on`/`ctx.effect`，避免裸订阅
换驱动后行为"丢了"	自定义 loop 退出了产品编排；确认你不需要预设/checkpoint/计费/重放

11. 完整示例：多角色量化研究 Agent

tutorials/agent-demo 是本教程的配套 demo：把 personas/ 下的 4 个角色定义（planner / executor / reviewer / synthesizer）做成一个真正可运行的 agent。编排是"一个 coordinator + 4 个角色子代理"——coordinator（main agent）的 persona 描述"计划 → 执行 → 审查 → 综合"流水线，模型通过 delegate_* 工具把每个阶段委派给对应角色，并把上一阶段的输出传给下一阶段。

每个角色 = 一个具名 ctx.subagents provider，核心代码只有一个类（agent-demo/src/providers.ts）：

export class RoleSubagentProvider implements SubagentProvider {
  readonly capabilities: SubagentCapabilities = {
    outputSchema: true, depthLimit: true, toolFilter: true, persona: true,
  }
  readonly inheritsParentContext = false // fresh child：零父上下文

  constructor(
    readonly name: RoleId,
    private readonly persona: string,
  ) {}

  start(request: ResolvedSubagentStartRequest): Promise<SubagentRun> {
    return startInProcessRun({ ...request, persona: this.persona }, {})
  }
}

start() 复用共享驱动 startInProcessRun()，在请求上盖 persona: <角色定义文本>，子代理即"这个角色"。插件入口（agent-demo/src/index.ts）只做注册：

export function apply(ctx: Context, config: Config) {
  const personas = loadRolePersonas(config.personasDir ?? defaultPersonasDir())
  for (const role of ROLES) {
    ctx.subagents.registerProvider(new RoleSubagentProvider(role, personas[role]))
  }
}

角色 persona 文件缺失时 loadRolePersonas() 加载即失败（fail loud）。剩下的全是组合（agent-demo/cordis.yml，第 4 章的路子）：dsh-agent-spine-demo 建 coordinator 并注入流水线 persona，dsh-llm-deepseek / dsh-bash-local / 会话持久化补齐执行环境，dsh-subagent 提供注册表，再用 4 个 dsh-tool-subagent 实例把角色暴露成模型可见的 delegate_planner / delegate_executor / delegate_reviewer / delegate_synthesizer 工具（maxDepth: 1，角色子代理不能再往下委派）。

跑起来（tutorials/ 不是 workspace 成员，先打包安装进 profile）：

cd tutorials/agent-demo && pnpm run build && pnpm pack
npx @deepseek-ai/dsh plugin --profile headless add ./deepseek-ai-dsh-agent-demo-0.1.0.tgz
export DEEPSEEK_API_KEY=...
npx @deepseek-ai/dsh --profile headless "分析贵州茅台：先规划，再执行，审查后综合成报告"

模型会依次调用 4 个 delegate_* 工具，只把最终综合报告返回给用户。完整说明（配置、测试、安装到 web profile、已知设计取舍）见 agent-demo/README.md。

DeepSeek Harness (`dsh`) 使用教程

Henry Lin — Tue, 18 Aug 2026 09:03:28 +0000

DeepSeek Harness (`dsh`) 使用教程

本教程面向想要快速上手 DeepSeek Harness 的开发者，覆盖从安装、配置到日常使用与扩展的全流程。

1. 项目简介

DeepSeek Harness（dsh）是 DeepSeek AI 开发的开源 agent harness（智能体框架），采用一切皆插件的架构，由 Cordis 驱动。

核心特点：

一切皆插件：包括模型适配器、工具注册表、会话日志、agent 主循环本身都是插件，任何部分都可以从配置层面替换。
能力缝（Capability Seam）：每个可替换能力由 Service Definition / Provider / Consumer 三角色组成，切换 Provider 即可整体改变产品行为。
会话日志：追加式事件日志是模型上下文的来源，"模型可见 ⟺ 已记录"是一条运行时不变量。
开发者预览：当前处于快速迭代阶段，会有破坏兼容性的变更。

常用术语：

术语	含义
Profile	命名的插件组合，存放在 `$DSH_HOME/profiles/<name>`
Bundle	插件的分发格式，由一组 Cordis 配置行 + 对应代码组成
Step	一次模型请求 + 它调用的工具
Turn	零到多个 step 组成的完整回合
Seam	可替换能力缝（见上）

2. 环境准备与安装

2.1 前置要求

Node.js：要求 ^22.19 或 >=24
pnpm（从源码运行时需要）
DeepSeek API Key（或其他兼容模型的 API Key）

2.2 方式一：通过 npm 直接运行（推荐体验）

不需要克隆仓库，只需安装 Node.js，然后：

npx @deepseek-ai/dsh web

该命令会启动 Web UI，默认地址为 http://127.0.0.1:3080。

2.3 方式二：从源码运行（推荐开发）

git clone https://github.com/deepseek-ai/deepseek-harness.git
cd deepseek-harness
pnpm install
pnpm run build
pnpm dsh web

注意：生产运行需要先执行 pnpm run build 生成构建产物；pnpm dsh 通过 tsx 以 ESM 模式直接从源码启动。

2.4 常用开发命令

pnpm run test            # 单元测试
pnpm run test:e2e        # 真实 API 测试（无 DEEPSEEK_API_KEY 时自动跳过）
pnpm run typecheck       # 类型检查
pnpm run lint            # 代码检查
pnpm run build           # 构建产物

3. 运行 Web UI

3.1 启动

pnpm dsh web
# 或使用 npm 版本：
# npx @deepseek-ai/dsh web

启动后浏览器打开 http://127.0.0.1:3080。

3.2 常用启动参数

pnpm dsh web --port 8080            # 自定义端口
pnpm dsh web --host 0.0.0.0         # 绑定地址（注意：CLI 暂不支持 0.0.0.0，会报错）
pnpm dsh web --trusted-host demo.com  # 添加受信任的域名（可重复）
pnpm dsh web --patch ./extra.yml    # 应用额外的 patch 覆盖
pnpm dsh web --dump-config          # 查看实际组合的插件树而不启动
pnpm dsh web --help                 # 查看 web 应用自身的参数

3.3 工作流程概览

Web UI 的完整使用流程：

配置模型（填入 API Key）
选择工作区（指定一个项目目录）
创建会话
发送任务
观察 agent 执行、审批弹窗与结果

4. 配置模型

4.1 配置 DeepSeek（最快上手）

打开 Settings → Models，在 DeepSeek 卡片中输入 API Key 并保存。

Key 是只写的：保存后页面只显示脱敏后的描述，不会回显明文。
Key 存储在 $DSH_HOME/.credentials.yaml，设置中只保留凭证引用。
保存后立即生效，无需重启服务。

4.2 添加目录中的 Provider

点击 Add provider，选择如 Anthropic、OpenAI 等，输入对应 API Key 保存即可。目录内置了端点、协议和模型列表。

部分 Provider 需要原生认证（如 Bedrock 用 AWS 凭证、Azure 用 api-version 等），只填 API Key 无法生效。

4.3 添加自定义 Provider

点击 Add a custom provider，适用于公司网关、自托管服务器或目录中没有的 Provider。需要填写：

Provider ID（小写，永久，后续请求/会话/默认模型都引用它）
显示名称
Base URL
API 协议
凭证
至少一个模型

模型可以通过 Fetch available models 从 GET /models 端点自动拉取，也可以手动输入。

4.4 图片输入（vision 模型）

手动输入的模型默认按"仅文本"处理。若模型支持图片，需要在 $DSH_HOME/settings.yaml 中声明：

llm-pi-ai:
  providers:
    vision-gateway:
      apiKeyEnv: GATEWAY_API_KEY
      api: openai-completions
      baseURL: https://vision.example/v1
      models:
        - id: first-model
          input: [text, image]

4.5 选择默认模型

配置好的 Provider 会出现在模型选择器中。选中的模型会成为新会话的默认模型。已发送过请求的会话会保留其日志中记录的模型。

5. 选择工作区

在 Web UI 中点击 Choose workspace：

添加启动 dsh 时的项目目录并选中。
会话创建入口在选中工作区之前不可用。
dsh 进程默认把启动目录作为文件系统位置。

工作区决定了 agent 可以读写的文件范围（配合权限策略生效）。

6. 创建会话并运行任务

6.1 创建会话

选中工作区后，新建一个会话。默认使用 workspace-write 权限预设：bash 和文件系统写操作被限制在工作区与平台临时目录内；读操作、网络访问和进程可见性不受限制。

6.2 发送任务示例

在会话中发送类似以下任务：

总结这个仓库，并指出它的主要包。

agent 可以：

读写和编辑工作区文件（read / write / edit / glob / grep）
运行命令（bash）
搜索网络（web_search / web_fetch）
委派子任务（subagent）
维护计划与待办（plan / todo_write）
询问用户（ask_user_question）

6.3 审批机制

当操作超出当前权限策略需要批准时，Web UI 会弹出确认请求。属于"审批缝（approval seam）"，缺省在没有应答者时 fail-closed 为 unavailable。

6.4 实战操作示例

下面是一组可以直接粘贴到 Web 会话中的任务，并说明预期的 agent 行为。把 <...> 替换为你的实际内容。

示例 1：代码库探索

这个仓库有哪些主要包？用树状图展示 packages/ 的目录结构，并总结每个包组（core、shell、fs、web 等）各自负责什么。

预期行为：agent 依次调用 glob/bash（列目录）、read（读关键 README 和 package.json）、grep（搜索特定模式），最终输出一份结构总结。这类任务通常不需要审批。

示例 2：修改代码并验证

在 src/utils.ts 中新增一个 formatBytes(bytes) 函数，把字节数格式化为可读的 "12.5 MB"。然后给它写一个单元测试并运行 pnpm test 确认通过。

预期行为：agent 先 read 目标文件（read-before-write 策略），再 write/edit 修改，然后写测试、跑 bash。对工作区的写操作一般直接执行，跨出工作区的命令才会触发审批。

示例 3：排查失败测试

运行 pnpm run test，如果失败，诊断根因并修复，最后再次运行确认全部通过。

预期行为：agent 用 bash 运行测试，读到失败后结合 read（看测试与源码）、grep（找相关代码）定位问题，edit 修复，再跑一遍验证。适合演示 agent 的循环-修复-验证工作流。

示例 4：联网检索

搜索 DeepSeek Harness 最新版本发布说明，总结 3 个主要新特性。

预期行为：agent 调用 web_search 获取结果，再对命中链接 web_fetch 读取正文，最后总结。若当前模型未配置搜索 Provider，web_search 会返回错误，此时应检查 base bundle 的搜索配置。

示例 5：计划 + 待办清单

为"给项目添加 CI"制定一个计划，并用待办清单跟踪每一步。

预期行为：agent 先进入计划模式整理步骤（可能调用 ask_user_question 确认选择），然后持续调用 todo_write 维护一个带状态（in_progress / completed）的清单，UI 会实时渲染为复选框。

示例 6：需要确认的交互

我接下来要把日志格式从 JSON 改成文本。开始前先问我：是否同时更新文档？

预期行为：agent 调用 ask_user_question，Web UI 弹出问题卡片（带 id、选项），你的回答会回填到工具结果中，agent 据此继续。审批与提问是两条不同机制：提问走 userQuestions 缝，审批走 approval 缝。

示例 7：并行委派子任务

同时调研两件事：A) 这个项目用到的持久化方案，B) 它的权限预设。分别交给子 agent，完成后汇总成一份对比表。

预期行为：agent 通过 subagent 工具派生子会话并行调研（也可在后台运行，用 job_list/job_output 收集），完成后在父会话汇总。子会话的事件会进入各自的会话日志，可通过会话历史回溯。

示例 8：跨会话检索

在历史会话中搜索所有提到 "sandbox" 的内容，把相关的结论贴出来。

预期行为：agent 调用 session_event_search / session_search 在持久化的会话日志中检索，结果基于当前工作区授权返回。演示了"会话即数据库"的能力。

提示：示例 1、2、3 只用到基础工具，任何配置了模型的工作区都能跑通，最适合首次体验。

7. headless 命令行模式

headless 是一次性运行的 CLI 模式，没有服务器、没有浏览器 UI。

7.1 运行一次性任务

pnpm dsh --profile headless "运行测试"

行为：

创建一个全新的持久化会话
提交任务，等待静默
刷新 Session，输出最后一条非空 assistant 文本
completed 时退出码 0，否则退出码 1
成功时 stdout 输出答案，stderr 无输出，不开放监听端口

7.2 headless 模式的特点

不挂载 ApiProxy、Host、HTTP 服务器、Web 运行时或浏览器客户端
没有任务文本属于用法错误
会话内容持久化，可从日志重建

8. 配置与权限

8.1 环境变量

变量	作用
`DEEPSEEK_API_KEY`	DeepSeek API Key（搜索也用此 key）
`DEEPSEEK_BASE_URL`	自定义 DeepSeek API 地址
`DEEPSEEK_SEARCH_BASE_URL`	自定义搜索 API 地址
`DSH_HOME`	Harness 数据目录（profiles、credentials、settings）
`DSH_PERMISSION_MODE`	改变进程的权限 fallback
`DSH_TOOLS_MODE`	工具模式：`native`、`code` 或 `both`
`DSH_TELEMETRY_MODE`	`FULL`（全量 OTLP 日志）或 `FEEDBACK_ONLY`
`DSH_TELEMETRY_OTLP_URL`	自定义遥测收集器地址
`DSH_TELEMETRY_DISABLED`	非空则彻底关闭遥测

8.2 凭证解析顺序

Provider 凭证按以下顺序解析（先到先得）：

继承的环境变量
$DSH_HOME/.credentials.yaml
启动目录的 .env
$DSH_HOME/.env

8.3 权限预设

预设	说明
`workspace-write`（默认）	bash/文件系统写操作限在工作区与临时根目录
`danger-full-access`	放开沙箱与审批限制

8.4 Profile 与 Patch 层

插件树的组合顺序（后层覆盖前层）：

profile 清单 dsh.profile.bundles 中列出的各 bundle（按顺序）
profile 的 cordis.patch.yml
家目录级 $DSH_HOME/cordis.patch.yml
--patch <path> 覆盖（按 argv 顺序）

patch 按行 id 定位，替换该行的整个 config 值（不是深合并），也可以插入新行。

查看实际组合的树：

pnpm dsh --profile web --dump-config
pnpm dsh --profile web --dump-default-config   # 只看 bundle 层

9. 插件管理

9.1 安装插件

pnpm dsh plugin --profile web add <package-or-git-spec>

转发给 pnpm 在 profile 目录执行
add、remove、why、update 等 pnpm 动词都可用
相对路径（.、../plugin）锚定到调用目录
依赖中声明了 "dsh": { "bundle": ... } 的包会自动进入 bundle 层

示例：

pnpm dsh plugin --profile tui add github:deepseek-harness/turtle-ui
pnpm dsh plugin --profile tui remove turtle-ui
pnpm dsh --profile tui

Git 托管的插件源码在安装时通过 prepare 脚本构建，pnpm ≥10 需要先在 profile 的 pnpm-workspace.yaml 中允许构建（按报错提示复制 allowBuilds key）。

9.2 内置 bundle

Bundle	作用
`@deepseek-ai/dsh-base`	每个 profile 的第一层：模型适配器、工具、持久化、沙箱、审批、设置、凭证、遥测
`@deepseek-ai/dsh-web-app`	浏览器应用（Web UI）
`@deepseek-ai/dsh-headless`	一次性运行器（无服务器）

web 和 headless profile 首次使用时会从内置模板自动初始化；其他 profile 需通过 dsh plugin 创建。

10. Python SDK

项目还提供 Python SDK，用于以编程方式驱动 dsh。详见 Python SDK 指南和 python/README.md。

10.1 安装

cd python/sdk
pip install -e .

10.2 基本用法

from dsh import DSHClient

client = DSHClient("http://127.0.0.1:3080")
session = client.create_session()

for event in session.send("总结这个仓库"):
    print(event)

具体 API 以 python/sdk/README.md 为准。

11. 开发插件（扩展能力）

11.1 扩展点速查

目标	机制
添加模型 Provider	在 `ctx.llm` 上注册适配器
添加模型可见能力（工具）	在 `ctx.tools` 上注册，schema 加入 prompt 组装
添加 bash 执行	注册 `ctx.shell` 后端
添加文件系统访问/策略	注册 `ctx.fs` Provider 或监听 `fs/*` 事件
添加持久终端执行	注册 `ctx.terminals` 后端 + `dsh-tool-terminal`
添加人工命令	在 `ctx.commands` 上注册
拦截请求/工具/回合	使用 `agent/` 或 `tools/` 事件
添加模型可见上下文	调用 `agent.inject()`
添加持久会话状态	扩展 `SessionEventMap`，从日志渲染与回放

11.2 快速开始

推荐从示例和 cookbook 开始：

开发指南：docs/development.md
架构文档：docs/architecture.md
扩展 cookbook：docs/cookbook/extension-cookbook.md
添加工具：docs/cookbook/adding-a-tool.md
添加 LLM 适配器：docs/cookbook/adding-an-llm-adapter.md
可运行示例：examples/

11.3 测试与质量门

pnpm run test              # 单元测试
pnpm run test:coverage     # CI 覆盖率门槛（每文件 100%）
pnpm run test:snapshot     # 无 key 的 ACP/headless 回放快照
pnpm run typecheck && pnpm run lint

11.4 插件发布

为插件仓库添加 dsh-plugin GitHub 话题便于被发现。发布流程见发布指南。

12. 常见问题排查

问题	解决方法
`MISSING_CREDENTIAL`	通过 Models 页保存 Provider Key，或提供引用的环境变量
`UNKNOWN_MODEL`	选择已配置的模型，或向自定义 Provider 添加缺失的模型
Fetch models 返回 401	检查 Key；端点无 `GET /models` 时手动输入模型
图片在发送前被拒绝	模型未声明图片模态；给自定义模型加 `input: [text, image]`
Git 插件 add 失败	在 profile 的 `pnpm-workspace.yaml` 中允许 `allowBuilds`
没有任务文本的 headless	headless 需要位置参数任务文本，空参数是用法错误
`dsh web --host 0.0.0.0` 报错	CLI 暂不支持 `0.0.0.0`，用 `--trusted-host` 添加受信任域名

DeepSeek Harness (`dsh`) 插件开发教程

Henry Lin — Tue, 18 Aug 2026 08:07:18 +0000

DeepSeek Harness (`dsh`) 插件开发教程

本教程面向想要为 dsh 编写插件的开发者，从概念到可运行的插件逐步走通。你将学会：认识"一切皆插件"的架构、写出函数插件、把它挂载到 dsh、做一个模型可见的工具插件、做一个拦截事件的钩子插件，最后了解如何做成正式包并过质量门。

核心概念：一切皆插件
环境准备
编写第一个插件
插件的四个导出：name / inject / Config / apply
把插件挂载到 dsh
实战一：模型可见的工具插件
实战二：拦截事件的钩子插件
声明式配置（Config）
能力缝：Service Definition / Provider / Consumer
仓库内开发一个正式包
测试与质量门
常见问题排查

1. 核心概念：一切皆插件

dsh 由 vendored 的 Cordis 驱动。产品里没有任何特权核心：模型适配器、工具注册表、会话日志、甚至 agent 主循环本身都是插件，因此任何部分都可以从配置层面替换或扩展。整个产品就是启动时从若干层组合出来的一棵插件树。

五个必须记住的想法（详见 docs/cordis-primer.md）：

想法	含义
插件	一个实现了 Service 的对象：最常见的是带 `apply(ctx)` 的函数，也可以是有 `inject` 的对象或 `Service` 子类
上下文（Context）	服务的仓库。服务挂到稳定的 `ctx.<key>`（如 `ctx.tools`、`ctx.llm`、`ctx.sessions`），插件之间通过 key 找服务，不 import 具体实现
inject	声明插件需要的服务。loader 会等待这些服务存在后再执行插件，加载顺序由依赖决定而不是文件顺序
类型化事件	服务通过声明合并定义事件，用 emit / waterfall / parallel / serial 分发给监听者
可逆的注册	工具 schema、prompt 段、适配器、监听器都通过 `ctx.effect()` / `ctx.on()` 注册；插件卸载（HMR、热重载、关停）时一切自动回滚

扩展点（event / 服务）是 dsh 的"API"。改行为时优先挂在扩展点上，不要改 loop。完整的对应表见 docs/architecture.md 的 Where new behavior goes。

2. 环境准备

推荐在一个 clone 下来的仓库里开发，这样能直接使用仓库的命令行和脚本。

git clone https://github.com/deepseek-ai/deepseek-harness.git
cd deepseek-harness
pnpm install

日常命令（docs/development.md）：

pnpm run typecheck     # 类型检查（strict，无 any 逃逸）
pnpm run lint          # oxlint
pnpm run test          # vitest 单元测试
pnpm run build         # tsc 产出 lib/types + tsdown 产出 lib/
pnpm dsh --profile headless "你好"   # 从源码跑一个一次性任务（需要 DEEPSEEK_API_KEY）

运行真实模型需要 API Key。把 DEEPSEEK_API_KEY 放进根目录 .env，pnpm dsh 会自动加载。

注意：本教程第 3 章的"纯链路演示"不需要 Key；第 6 章起的工具/钩子插件要真正跑模型时需要 Key。没有 Key 也可以先写代码、跑单元测试和 dump-config 验证。

3. 编写第一个插件

3.1 最小函数插件

在任意开发目录（这里假设仓库根下的 tmp/hello-plugin/）创建 hello.ts：

import type { Context } from '@deepseek-ai/cordis'

export const name = 'hello'

export function apply(ctx: Context) {
  ctx.logger.info('hello from my first plugin')
}

name 是可选显示元数据，用于诊断信息中标识插件。apply(ctx) 是插件贡献一切的地方。

再创建 cordis.yml：

- name: './hello.ts'

cordis.yml 是一组 Cordis 配置项列表，name 可以是相对路径或 npm 包名。loader 会挂载每一条。

3.2 用 vendored loader 跑起来（不需要 Key）

仓库内带了 vendored 的 Cordis 启动器，可以直接走一遍最小的挂载链路：

node --import tsx ../../vendor/cordis/bin.js

预期输出：

[info] hello from my first plugin

发生了什么：

启动器创建根 Context 并挂载 Loader 插件。
Loader 读取 cordis.yml，解析 ./hello.ts 并作为子插件挂载。
Cordis 调用你的 apply(ctx)。

日志导出器装好后进程会一直等待事件，没有其他事件时 Ctrl-C 退出即可。这个例子摘取自正式的 Cordis 教程第 1 章，那里有更完整的讲解。

3.3 插件不只是函数

Cordis 接受三种形态；在你需要公开服务之前，一直用函数形态：

import { Service, type Context } from '@deepseek-ai/cordis'

// 1. 函数插件（最常见）
export function apply(ctx: Context) {}

// 2. 对象插件：带 apply 方法的对象
export const objectPlugin = {
  name: 'object-plugin',
  apply(ctx: Context) {},
}

// 3. 类插件：Service 子类（适合对外提供一个 ctx.<key> 服务）
export class MyService extends Service {
  constructor(ctx: Context) {
    super(ctx, 'myService')
  }
}

4. 插件的四个导出：name / inject / Config / apply

一个正式的函数插件通常导出四个东西：

import type { Context } from '@deepseek-ai/cordis'
import z from '@deepseek-ai/schemastery'

/** 插件显示名，仅用于诊断。 */
export const name = 'my-plugin'

/** 声明依赖的必需服务；loader 会等它们存在再执行 apply。 */
export const inject = ['tools']

/** 部署期配置的 schemastery 校验 schema（可省略，见第 8 章）。 */
export interface Config {
  greeting: string
}
export const Config: z<Config> = z.object({
  greeting: z.string(),
})

/** 插件主体：注册一切贡献，并只注册为可逆 effect。 */
export function apply(ctx: Context, config: Config) {
  ctx.logger.info(config.greeting)
}

要点：

inject 只声明必需服务。可选项用 ctx.get(name) 读取全局服务仓库，不要用属性代理（拓扑敏感）。
function 插件必须命名导出，不要配默认导出。混合两种形式会让 Loader 丢掉 inject 元数据（见 packages/AGENTS.md 的 postmortem 引用）。
apply 签名：有 Config 导出时是 (ctx, config)，没有时是 (ctx)。
误配置要 fail loud：加载失败会明确报错，不会静默跳过。

5. 把插件挂载到 dsh

5.1 插件树怎么来的

一个运行中的 dsh 由若干层组合而成，后层覆盖前层：

profile 清单里列出的各 bundle（顺序加载）
profile 的 cordis.patch.yml
家目录级 $DSH_HOME/cordis.patch.yml
--patch <path> 覆盖（按 argv 顺序）

查看你机器实际组合出的树：

pnpm dsh --profile headless --dump-config

任何打印出来的行，都可以用你自己的 patch 替换。patch 按行 id 定位：要么替换那行的整个 config（不是深合并），要么插入新行。

# overlay.yml
- id: my-tool
  name: '@deepseek-ai/dsh-tool-mine'
  config:
    option: value

5.2 三条挂载路径

路径	适用	做法
外置插件（推荐大多数场景）	自研、不开源、单独发布	独立 npm 包，profile 里 `dsh plugin add` 安装；package.json 声明 `"dsh": {"bundle": ...}` 可自动进 bundle 层
临时 overlay	调试、演示	`pnpm dsh --profile <name> --patch ./overlay.yml "任务"`
仓库内包	给 dsh 本身贡献代码	放 `packages/<group>/<pkg>`，见第 10 章

5.3 安装外置插件

dsh plugin 子命令在 profile 目录里调用 pnpm，动词直接透传：

pnpm dsh plugin --profile headless add /path/to/my-plugin
pnpm dsh plugin --profile headless add github:some-org/my-plugin
pnpm dsh plugin --profile headless remove my-plugin

相对路径（.、../plugin）锚定到命令行所在目录。
包依赖里声明了 "dsh": { "bundle": ... } 的，自动进入该 profile 的 bundle 层。
Git 托管的插件用 prepare 脚本构建；pnpm ≥10 需要在 profile 的 pnpm-workspace.yaml 里允许构建（按报错提示复制 allowBuilds key）。

6. 实战一：模型可见的工具插件

插件最常见的用途是给模型加工具。工具注册在 ctx.tools 上，schema 会自动进入 prompt 组装，模型就能"看到"它。

下面是一个完整可运行的最小工具（省略 import）：

import { readFile } from 'node:fs/promises'
import type { Context } from '@deepseek-ai/cordis'
import { defineTool } from '@deepseek-ai/dsh-tools'

export const name = 'demo-tool'
export const inject = ['tools']

export function apply(ctx: Context) {
  ctx.tools.register(defineTool({
    name: 'read_file',
    description: 'Read a file from disk.',       // 模型看到的能力描述
    parameters: {
      path: { type: 'string', required: true, description: 'Absolute path' },
      limit: { type: 'number' },                  // 可选项，默认不要求提供
    },
    output: {
      schema: { type: 'string' },
      render: (_args, value) => [{ type: 'text', text: value }],
    },
    async execute(args, exec) {
      // args 已经被 defineTool 按 schema 校验并推导了类型
      return readFile(args.path, { encoding: 'utf8', signal: exec.signal })
    },
  }))
}

execute() 契约的几条硬规则（完整版见 docs/cookbook/adding-a-tool.md）：

args 自动校验：defineTool 会在 execute 前校验模型生成的参数，execute 里拿到的 args 类型和 schema 一致。
只返回一个规范 JSON 值：output.schema 定义返回值；execute 只返回这个值。抛异常 = isError；成功的领域结果（如非零退出码）也要放进规范值返回。
遵守 exec.signal：信号触发时要取消进行中的工作。
只注册一次：注册借用的是只有读权限的 definition，不要事后改 schema；想换工具就释放它所属的 effect 再注册。
UI 卡片是独立设计：模型看到的内容由 output.render 决定；UI 卡片由 presentCall / presentResult 返回 generic / terminal / diff 渲染意图。

6.1 后台长任务

需要 run_in_background 且由部署配置开关（不能写死在代码里）的工具，通过 ctx.jobs.start() 注册为后台任务更合理，模型侧返回带 jobId 的规范句柄，job_output / job_kill 负责收集与停止。见 adding-a-tool.md 的 Long-running work。

6.2 工具如何渲染成 UI

选中 card 标签的渲染意图，按类型给卡片：

{ card: 'generic', title, locations? } —— 通用卡片；locations: [{ path, line? }] 让编辑器跳转工具碰过的文件。
{ card: 'terminal', title, cwd? } —— 你的调用就是 shell 命令（参考 dsh-tool-bash）。
{ card: 'diff', title, diffs } —— 创建/修改文件，内联 diff 卡片（参考 dsh-tool-fs 的 write/edit）。

7. 实战二：拦截事件的钩子插件

不需要新工具，只想在某个环节插一脚时，用事件监听器。主循环通过事件驱动，钩子插件就是在这些事件上挂监听器。

7.1 一个权限门示例

下面的插件在 tools/pre-execute 上拦截每一次工具调用，按规则允许或拒绝：

import type { Context } from '@deepseek-ai/cordis'
import type { PreToolDecision, ToolExecution } from '@deepseek-ai/dsh-tools'

declare function isAllowed(exec: ToolExecution): Promise<boolean>

export const name = 'permission-gate'

export function apply(ctx: Context) {
  ctx.on('tools/pre-execute', async (exec, next): Promise<PreToolDecision> => {
    if (!(await isAllowed(exec))) {
      return { kind: 'deny', reason: 'Denied by policy.' }
    }
    return next()
  })
}

tools/pre-execute 是 waterfall 事件：监听器收到 (...args, next)，调用 next() 把结果传给下一个监听器；不调 next() 直接 return 就是短路（截断整条链）。这是 Cordis waterfall 的核心语义，写监听器时最容易踩的坑就是忘了 next()。

7.2 挑选正确的扩展点

你要做的	用哪个
允许 / 拒绝 / 询问工具调用	`tools/pre-execute`，返回 `{kind:'deny'}` / `{kind:'ask'}`
工具调用必须被最终否决、不可被撤销	`ctx.tools.guard()`
包裹工具执行生命周期（超时/重试/指标）	`tools/execute`（只有 `exec.signal` 可替换）
显式改写工具结果或呈现内容	`tools/post-execute`
只观察不可变最终结果（审计/捕获）	`tools/result`
改写模型请求配置	`agent/request`（waterfall，返回替换后的 `LlmCallConfig`）
改写/拒绝进入 step 的消息	`agent/pre-step`（waterfall，返回 `PreStepDecision`）
收尾 turn / 强制再走一步	`agent/turn-stopping`（serial，无 `next()`）
监听最终不可变输出（工具结果、状态、error）	对应 emit 事件 + `session/event` 上的 durable 会话事件

事件的生产者 / 消费者完整清单见 docs/event-producer-consumer.md。

7.3 durable 事件和 live 事件的分工

durable 会话事件（turn/*、step/*、user/message、assistant/*、tool/*）追加进会话日志，重启后仍可重建。模型能看到的任何东西都必须能从这个日志重建（模型可见 ⟺ 已记录）。
live 事件（agent/*、tools/*）只做运行期协调：queue/status、拦截、请求构造、steering。

要给模型加新的可见输入，就遵从这条规则：往扩展 SessionEventMap 里加一种新事件类型，从日志渲染，而不是绕过日志。

8. 声明式配置（Config）

部署期可调的选择不能写死在插件里，而是做成 Config 字段，从 cordis.yml 修改。

8.1 定义 schema

用 @deepseek-ai/schemastery（它也是 Cordis 的校验器），类型和运行时校验合一：

import z from '@deepseek-ai/schemastery'

export interface Config {
  allowParallelInProgress: boolean
}

export const Config: z<Config> = z.object({
  allowParallelInProgress: z.boolean().required(),
})

8.2 在 cordis.yml 里装配

- id: todo
  name: '@deepseek-ai/dsh-tool-todo'
  config:
    allowParallelInProgress: true

8.3 配置只是 metadata，条件组合用 overlay

cordis.yml 的 !!js 只允许出现在插件 config 和条目 disabled 下；其他 metadata 保持字面量（loader 语义）。
按环境选插件用 overlay，不要用 !js。

# patch 里按环境取值是合法用法
- id: sandbox-policy
  name: '@deepseek-ai/dsh-sandbox-policy'
  config:
    mode: !!js "process.env.DSH_PERMISSION_MODE ?? 'workspace-write'"

9. 能力缝：Service Definition / Provider / Consumer

当一个能力需要"可替换"时，dsh 用的不是一个类，而是缝（seam）：三个角色。

角色	职责	例子
Service Definition	声明接口（服务 + 事件词汇），挂在 `ctx.<key>`	`dsh-subagent` 提供 `ctx.subagents` 和一次调用词汇
Service Provider	实现这个接口	`dsh-subagent-spawn-in-process`、`-fork`、`-acp`、`-codex`…
Consumer	消费它，通常是模型可用的工具	`dsh-tool-subagent` 把配置好的 provider 暴露给模型

换一个 Provider 就整体改变产品行为：文件系统和子进程 Provider 共享同一执行世界，把 ctx.shell 指向远程沙箱，bash / PTY / LSP 一起跟着换，不用派生 Provider（见 docs/architecture.md 的 Capability seams）。

一个角色不算缝。要新增能力，就设计全部三个角色——即使开始时 Provider 只有一个。经典模板是 shell 三件套，代码在 packages/shell/。

10. 仓库内开发一个正式包

如果你的插件要成为 @deepseek-ai/dsh-<name> 的一部分，按 checklist 建包。完整文件级清单见 docs/cookbook/adding-a-package.md，这里说核心：

packages/<group>/<pkg>/
  package.json     # private: true；version 与根 package.json 一致；type: module
                   # main/types/exports 指向 lib/；@deepseek-ai/cordis 在 peer+dev deps
  tsconfig.json    # extends ../../../tsconfig.base.json；references 指向依赖的 workspace 包
  src/index.ts     # 函数插件命名导出 name/inject/Config/apply；服务类默认导出类
  README.md        # 服务 API、事件、扩展点、Model Experience 段

流程：

选组：core、llm、shell、fs、web 等已有组匹配能力角色就放进组的 packages/<group>/<pkg>。
注册：tsconfig.host.json / tsconfig.client.json 的 references 加一行；新组才动 tsconfig.base.json。
命名：接口包命名能力（dsh-subagent），实现包加机制（-in-process、-fork）。ctx key 单数表示一个引擎，复数表示注册表。
验证：跑下面的质量门。

11. 测试与质量门

11.1 本地必跑

新增/修改包后，逐级往上跑（尽量只跑影响的，CI 才全量；详见 docs/testing.md）：

pnpm run constraints   # workspace 约束
pnpm run typecheck
pnpm run lint
pnpm run build
pnpm run hygiene       # knip + publint + NodeNext 消费检查
pnpm run doc-sync      # 文档门
pnpm run test          # 单元测试

11.2 测试方针

行为测试描述行为，不计正确性；改行为要改测试并说明原因。
每文件 100% 覆盖率是 CI 门槛（pnpm run test:coverage）。
产品可见的插件要有一个真实组合测试：通过 Loader 和 app/process 启动 cordis.yml，而不是只用手搭的 ctx.plugin(...) 单测。
每个非平凡、模型可见 / 产品用户可见的行为变化，都要在同一个 PR 里补一个可回放的 keyless 快照（pnpm run test:snapshot）。
注册可逆性用 HMR 安全测试验证：释放 fiber 后注册消失。

11.3 文档义务

每个非平凡改动带一篇 Agent Note（.agents/notes/）。
改行为（config key、默认值、错误码、wire 字段）要和 README / JSDoc 同 commit 更新。
包 README 要按规范写 Model Experience 段和 Known Limitations 段。

12. 常见问题排查

现象	原因 / 处理
插件没生效	先查 `name` 拼写：模块解析失败只记日志不崩溃，且启动早期可能丢失。再 `--dump-config` 确认真实组合树里有没有它
apply 抛异常	插件加载失败会明确终止进程，是设计如此，不会静默跳过
新配置项没效果	patch 按 id 替换整个 `config`，不是深合并；检查是否覆盖丢了其他字段
`!!js` 报错	`!!js` 只允许在插件 `config` 和条目 `disabled` 下，别处要用 overlay
模型看不到我的工具	工具 schema 自动进 prompt，但 scope（预设隔离）或 `restrict()` 可能把它滤掉了
忘记 `next()`	waterfall 监听器不调 `next()` 会短路整条链；要放行必须调用
HMR 热重载后残留	注册必须是 `ctx.effect()` / `ctx.on()` 系，`apply` 里裸干的事不会被回滚
`dsh plugin add` 构建失败	pnpm ≥10 需要在 profile 的 `pnpm-workspace.yaml` 允许 `allowBuilds`

NautilusTrader第8章：订单管理系统

Henry Lin — Thu, 16 Jul 2026 22:40:41 +0000

第8章：订单管理系统

学习目标

通过本章学习，您将：

掌握各种订单类型的使用方法
理解订单生命周期和状态管理
学会高级订单策略（OCO、OTO等）
实现智能订单管理
掌握订单执行和风险管理

8.1 订单类型概览

8.1.1 基础订单类型

NautilusTrader 支持所有常见的订单类型：

"""
基础订单类型示例
"""

from decimal import Decimal
from nautilus_trader.model.orders import (
    MarketOrder,
    LimitOrder,
    StopMarketOrder,
    StopLimitOrder,
)
from nautilus_trader.model.enums import (
    OrderSide,
    TimeInForce,
    OrderStatus,
)
from nautilus_trader.model.objects import Price, Quantity
from nautilus_trader.model.identifiers import (
    InstrumentId,
    OrderId,
    StrategyId,
    TraderId,
)
from nautilus_trader.common.enums import LogColor
from nautilus_trader.trading.strategy import Strategy


class OrderTypesExample:
    """订单类型示例类"""

    def __init__(self, trader_id: TraderId, strategy_id: StrategyId):
        """初始化"""
        self.trader_id = trader_id
        self.strategy_id = strategy_id
        self.instrument_id = InstrumentId.from_str("BTCUSDT.BINANCE")

    def create_market_order(self, side: OrderSide, quantity: Decimal) -> MarketOrder:
        """创建市价单"""
        return MarketOrder(
            trader_id=self.trader_id,
            strategy_id=self.strategy_id,
            instrument_id=self.instrument_id,
            order_side=side,
            quantity=Quantity(quantity),
            time_in_force=TimeInForce.GTC,  # Good Till Cancelled
            post_only=False,  # 允许吃单
            reduce_only=False,  # 非减仓单
        )

    def create_limit_order(
        self,
        side: OrderSide,
        quantity: Decimal,
        price: Price,
        **kwargs
    ) -> LimitOrder:
        """创建限价单"""
        # 默认参数
        default_kwargs = {
            'time_in_force': TimeInForce.GTC,
            'post_only': False,
            'reduce_only': False,
            'display_qty': None,
            'emulation_trigger': None,
            'trigger_instruction': None,
            'expire_time': None,
        }

        # 合并自定义参数
        default_kwargs.update(kwargs)

        return LimitOrder(
            trader_id=self.trader_id,
            strategy_id=self.strategy_id,
            instrument_id=self.instrument_id,
            order_side=side,
            quantity=Quantity(quantity),
            price=price,
            **default_kwargs
        )

    def create_stop_market_order(
        self,
        side: OrderSide,
        quantity: Decimal,
        trigger_price: Price,
        **kwargs
    ) -> StopMarketOrder:
        """创建止损市价单"""
        return StopMarketOrder(
            trader_id=self.trader_id,
            strategy_id=self.strategy_id,
            instrument_id=self.instrument_id,
            order_side=side,
            quantity=Quantity(quantity),
            trigger_price=trigger_price,
            time_in_force=TimeInForce.GTC,
            reduce_only=True,  # 止损单通常是减仓
            **kwargs
        )

    def create_stop_limit_order(
        self,
        side: OrderSide,
        quantity: Decimal,
        trigger_price: Price,
        limit_price: Price,
        **kwargs
    ) -> StopLimitOrder:
        """创建止损限价单"""
        return StopLimitOrder(
            trader_id=self.trader_id,
            strategy_id=self.strategy_id,
            instrument_id=self.instrument_id,
            order_side=side,
            quantity=Quantity(quantity),
            trigger_price=trigger_price,
            price=limit_price,
            time_in_force=TimeInForce.GTC,
            reduce_only=True,
            **kwargs
        )


# 订单使用示例
class OrderSubmissionExample(Strategy):
    """订单提交示例策略"""

    def __init__(self):
        """初始化策略"""
        super().__init__()
        self.example_orders = OrderTypesExample(self.trader_id, self.id)
        self.submitted_orders = {}

    def on_start(self):
        """策略启动"""
        self.log.info("订单提交示例策略启动")
        self.subscribe_bars(self.bar_type)

    def on_bar(self, bar: Bar):
        """处理K线数据"""
        if self.count_bars() % 10 == 0:  # 每10根K线测试不同订单类型
            self._test_order_types(bar)

    def _test_order_types(self, bar: Bar):
        """测试不同订单类型"""
        current_price = bar.close

        # 1. 市价单买入
        if self.count_bars() % 50 == 0:
            market_order = self.example_orders.create_market_order(
                OrderSide.BUY,
                Decimal("0.01")
            )
            self.submit_order(market_order)
            self.log.info("提交市价买单", color=LogColor.BLUE)

        # 2. 限价单卖出
        elif self.count_bars() % 30 == 0:
            limit_price = Price.from_str(str(current_price * Decimal("1.001")))
            limit_order = self.example_orders.create_limit_order(
                OrderSide.SELL,
                Decimal("0.01"),
                limit_price
            )
            self.submit_order(limit_order)
            self.log.info(f"提交限价卖单 - 价格: {limit_price}", color=LogColor.ORANGE)

        # 3. 止损单
        elif self.count_bars() % 20 == 0:
            stop_price = Price.from_str(str(current_price * Decimal("0.99")))
            stop_order = self.example_orders.create_stop_market_order(
                OrderSide.SELL,
                Decimal("0.01"),
                stop_price
            )
            self.submit_order(stop_order)
            self.log.info(f"提交止损单 - 触发价: {stop_price}", color=LogColor.RED)

8.1.2 高级订单类型

"""
高级订单类型示例
"""

from nautilus_trader.model.orders import (
    MarketIfTouchedOrder,
    LimitIfTouchedOrder,
    TrailingStopMarketOrder,
    TrailingStopLimitOrder,
)
from nautilus_trader.model.enums import (
    TriggerType,
    TrailingOffset,
)


class AdvancedOrderTypesExample(OrderTypesExample):
    """高级订单类型示例"""

    def create_market_if_touched_order(
        self,
        side: OrderSide,
        quantity: Decimal,
        trigger_price: Price,
        **kwargs
    ) -> MarketIfTouchedOrder:
        """创建市价触发单"""
        return MarketIfTouchedOrder(
            trader_id=self.trader_id,
            strategy_id=self.strategy_id,
            instrument_id=self.instrument_id,
            order_side=side,
            quantity=Quantity(quantity),
            trigger_price=trigger_price,
            trigger_type=TriggerType.DEFAULT,
            time_in_force=TimeInForce.GTC,
            **kwargs
        )

    def create_limit_if_touched_order(
        self,
        side: OrderSide,
        quantity: Decimal,
        trigger_price: Price,
        limit_price: Price,
        **kwargs
    ) -> LimitIfTouchedOrder:
        """创建限价触发单"""
        return LimitIfTouchedOrder(
            trader_id=self.trader_id,
            strategy_id=self.strategy_id,
            instrument_id=self.instrument_id,
            order_side=side,
            quantity=Quantity(quantity),
            trigger_price=trigger_price,
            price=limit_price,
            trigger_type=TriggerType.DEFAULT,
            time_in_force=TimeInForce.GTC,
            **kwargs
        )

    def create_trailing_stop_market_order(
        self,
        side: OrderSide,
        quantity: Decimal,
        trailing_offset: Decimal,
        **kwargs
    ) -> TrailingStopMarketOrder:
        """创建追踪止损市价单"""
        return TrailingStopMarketOrder(
            trader_id=self.trader_id,
            strategy_id=self.strategy_id,
            instrument_id=self.instrument_id,
            order_side=side,
            quantity=Quantity(quantity),
            trailing_offset=TrailingOffset(
                trailing_offset=trailing_offset,
                offset_type='PRICE',
            ),
            trigger_type=TriggerType.DEFAULT,
            time_in_force=TimeInForce.GTC,
            reduce_only=True,
            **kwargs
        )

    def create_trailing_stop_limit_order(
        self,
        side: OrderSide,
        quantity: Decimal,
        trailing_offset: Decimal,
        limit_offset: Decimal,
        **kwargs
    ) -> TrailingStopLimitOrder:
        """创建追踪止损限价单"""
        return TrailingStopLimitOrder(
            trader_id=self.trader_id,
            strategy_id=self.strategy_id,
            instrument_id=self.instrument_id,
            order_side=side,
            quantity=Quantity(quantity),
            trailing_offset=TrailingOffset(
                trailing_offset=trailing_offset,
                offset_type='PRICE',
            ),
            limit_offset=TrailingOffset(
                trailing_offset=limit_offset,
                offset_type='PRICE',
            ),
            trigger_type=TriggerType.DEFAULT,
            time_in_force=TimeInForce.GTC,
            reduce_only=True,
            **kwargs
        )


# 高级订单使用策略
class AdvancedOrderStrategy(Strategy):
    """高级订单策略"""

    def __init__(self):
        """初始化策略"""
        super().__init__()
        self.example_orders = AdvancedOrderTypesExample(self.trader_id, self.id)

        # 状态跟踪
        self.position_open = False
        self.entry_price = None
        self.trailing_order_id = None

    def on_start(self):
        """策略启动"""
        self.log.info("高级订单策略启动")
        self.subscribe_bars(self.bar_type)

    def on_bar(self, bar: Bar):
        """处理K线数据"""
        current_price = bar.close

        if not self.position_open:
            # 尝试开仓
            self._try_open_position(current_price)
        else:
            # 更新追踪止损
            self._update_trailing_stop(current_price)

    def _try_open_position(self, current_price: Decimal):
        """尝试开仓"""
        # 使用触发单
        trigger_price = Price.from_str(str(current_price * Decimal("1.001")))

        mit_order = self.example_orders.create_market_if_touched_order(
            OrderSide.BUY,
            Decimal("0.01"),
            trigger_price
        )

        self.submit_order(mit_order)
        self.log.info(
            f"提交MIT买单 - 触发价: {trigger_price}, "
            f"当前价: {current_price}",
            color=LogColor.GREEN
        )

    def on_order_filled(self, event):
        """处理订单成交"""
        if event.order.side == OrderSide.BUY and not self.position_open:
            # 开仓成功
            self.position_open = True
            self.entry_price = event.last_px
            self.log.info(f"开仓成功 - 价格: {self.entry_price}")

            # 设置追踪止损
            self._set_initial_trailing_stop()

    def _set_initial_trailing_stop(self):
        """设置初始追踪止损"""
        if self.position_open:
            # 2%的追踪距离
            trailing_offset = Decimal("0.02")
            current_price = self.cache.get_last_price(self.example_orders.instrument_id)

            if current_price:
                trailing_stop = self.example_orders.create_trailing_stop_market_order(
                    OrderSide.SELL,
                    Decimal("0.01"),
                    trailing_offset
                )

                self.submit_order(trailing_stop)
                self.trailing_order_id = trailing_stop.client_order_id
                self.log.info(
                    f"设置追踪止损 - 距离: {trailing_offset*100}%, "
                    f"当前价: {current_price}"
                )

    def _update_trailing_stop(self, current_price: Decimal):
        """更新追踪止损"""
        if self.position_open and self.trailing_order_id:
            # 检查是否需要调整追踪距离
            unrealized_pnl = self._calculate_unrealized_pnl(current_price)

            # 如果盈利超过3%，紧化追踪距离
            if unrealized_pnl > Decimal("0.03"):
                self._tighten_trailing_stop(current_price)

    def _tighten_trailing_stop(self, current_price: Decimal):
        """紧化追踪止损"""
        # 取消旧订单
        if self.trailing_order_id:
            self.cancel_order(self.trailing_order_id)
            self.log.info("取消旧追踪止损订单")

        # 设置新的追踪止损（1%距离）
        new_trailing_offset = Decimal("0.01")
        new_trailing_stop = self.example_orders.create_trailing_stop_market_order(
            OrderSide.SELL,
            Decimal("0.01"),
            new_trailing_offset
        )

        self.submit_order(new_trailing_stop)
        self.trailing_order_id = new_trailing_stop.client_order_id
        self.log.info(
            f"紧化追踪止损 - 新距离: {new_trailing_offset*100}%, "
            f"当前价: {current_price}"
        )

    def _calculate_unrealized_pnl(self, current_price: Decimal) -> Decimal:
        """计算未实现盈亏"""
        if not self.entry_price:
            return Decimal("0")

        return (current_price - self.entry_price) / self.entry_price

8.2 订单生命周期管理

8.2.1 订单状态和事件

"""
订单状态管理
"""

from nautilus_trader.model.events import (
    OrderInitialized,
    OrderSubmitted,
    OrderAccepted,
    OrderRejected,
    OrderCanceled,
    OrderExpired,
    OrderTriggered,
    OrderPendingUpdate,
    OrderPendingCancel,
    OrderFilled,
    OrderPartiallyFilled,
)
from typing import Dict, List, Optional
from datetime import datetime


class OrderLifecycleManager:
    """订单生命周期管理器"""

    def __init__(self):
        """初始化生命周期管理器"""
        self.order_states = {}  # order_id -> state info
        self.order_history = {}  # order_id -> event list
        self.active_orders = set()  # active order IDs
        self.filled_orders = {}  # order_id -> fill info

    def track_order(self, order_id: str):
        """开始追踪订单"""
        self.order_states[order_id] = {
            'created_at': datetime.utcnow(),
            'last_updated': datetime.utcnow(),
            'status': None,
            'filled_qty': Decimal('0'),
            'filled_amount': Decimal('0'),
            'commission': Decimal('0'),
        }
        self.order_history[order_id] = []
        self.active_orders.add(order_id)

    def update_order_state(self, event):
        """更新订单状态"""
        order_id = str(event.order_id)

        if order_id not in self.order_states:
            self.track_order(order_id)

        # 更新历史
        self.order_history[order_id].append({
            'timestamp': datetime.utcnow(),
            'event_type': type(event).__name__,
            'event': event,
        })

        # 更新状态
        state = self.order_states[order_id]
        state['last_updated'] = datetime.utcnow()

        # 处理特定事件
        if isinstance(event, OrderFilled):
            self._handle_fill(event, order_id, state)
        elif isinstance(event, OrderPartiallyFilled):
            self._handle_partial_fill(event, order_id, state)
        elif isinstance(event, OrderCanceled):
            self._handle_cancellation(event, order_id, state)
        elif isinstance(event, OrderRejected):
            self._handle_rejection(event, order_id, state)
        elif isinstance(event, OrderExpired):
            self._handle_expiry(event, order_id, state)

    def _handle_fill(self, event, order_id: str, state: dict):
        """处理完全成交"""
        state['status'] = 'FILLED'
        state['filled_qty'] = event.last_qty
        state['filled_amount'] = event.last_qty * event.last_px
        state['commission'] = event.commission
        state['filled_at'] = datetime.fromtimestamp(event.ts_event / 1e9)

        self.active_orders.remove(order_id)
        self.filled_orders[order_id] = {
            'order': event.order,
            'fill': event,
            'state': state.copy(),
        }

        self.log.info(f"订单完全成交: {order_id}")

    def _handle_partial_fill(self, event, order_id: str, state: dict):
        """处理部分成交"""
        state['status'] = 'PARTIALLY_FILLED'
        state['filled_qty'] += event.last_qty
        state['filled_amount'] += event.last_qty * event.last_px
        state['commission'] += event.commission

        self.log.info(f"订单部分成交: {order_id}, "
                     f"成交数量: {event.last_qty}, "
                     f"已成交总量: {state['filled_qty']}")

    def _handle_cancellation(self, event, order_id: str, state: dict):
        """处理取消"""
        state['status'] = 'CANCELED'
        state['canceled_at'] = datetime.fromtimestamp(event.ts_event / 1e9)

        if order_id in self.active_orders:
            self.active_orders.remove(order_id)

        self.log.info(f"订单已取消: {order_id}")

    def _handle_rejection(self, event, order_id: str, state: dict):
        """处理拒绝"""
        state['status'] = 'REJECTED'
        state['rejected_at'] = datetime.fromtimestamp(event.ts_event / 1e9)
        state['rejection_reason'] = event.reason

        if order_id in self.active_orders:
            self.active_orders.remove(order_id)

        self.log.warning(f"订单被拒绝: {order_id}, 原因: {event.reason}")

    def _handle_expiry(self, event, order_id: str, state: dict):
        """处理过期"""
        state['status'] = 'EXPIRED'
        state['expired_at'] = datetime.fromtimestamp(event.ts_event / 1e9)

        if order_id in self.active_orders:
            self.active_orders.remove(order_id)

        self.log.warning(f"订单已过期: {order_id}")

    def get_order_state(self, order_id: str) -> Optional[dict]:
        """获取订单状态"""
        return self.order_states.get(order_id)

    def get_active_orders(self) -> List[str]:
        """获取活跃订单"""
        return list(self.active_orders)

    def get_order_history(self, order_id: str) -> List[dict]:
        """获取订单历史"""
        return self.order_history.get(order_id, [])

    def get_order_stats(self) -> dict:
        """获取订单统计"""
        total_orders = len(self.order_states)
        active_count = len(self.active_orders)
        filled_count = len(self.filled_orders)

        # 计算成功率
        submitted_count = sum(1 for state in self.order_states.values()
                             if state['status'] in ['FILLED', 'CANCELED', 'REJECTED', 'EXPIRED'])

        success_rate = (filled_count / submitted_count) if submitted_count > 0 else 0

        return {
            'total_orders': total_orders,
            'active_orders': active_count,
            'filled_orders': filled_count,
            'success_rate': success_rate,
            'rejection_rate': sum(1 for state in self.order_states.values()
                                 if state['status'] == 'REJECTED') / total_orders if total_orders > 0 else 0,
        }


# 使用生命周期管理的策略
class LifecycleManagedStrategy(Strategy):
    """生命周期管理策略"""

    def __init__(self):
        """初始化策略"""
        super().__init__()
        self.lifecycle_manager = OrderLifecycleManager()

    def on_order_initialized(self, event: OrderInitialized):
        """处理订单初始化"""
        self.lifecycle_manager.update_order_state(event)

    def on_order_submitted(self, event: OrderSubmitted):
        """处理订单提交"""
        self.lifecycle_manager.update_order_state(event)

    def on_order_accepted(self, event: OrderAccepted):
        """处理订单接受"""
        self.lifecycle_manager.update_order_state(event)

    def on_order_rejected(self, event: OrderRejected):
        """处理订单拒绝"""
        self.lifecycle_manager.update_order_state(event)
        self._handle_rejection(event)

    def on_order_canceled(self, event: OrderCanceled):
        """处理订单取消"""
        self.lifecycle_manager.update_order_state(event)

    def on_order_expired(self, event: OrderExpired):
        """处理订单过期"""
        self.lifecycle_manager.update_order_state(event)

    def on_order_triggered(self, event: OrderTriggered):
        """处理订单触发"""
        self.lifecycle_manager.update_order_state(event)

    def on_order_pending_update(self, event: OrderPendingUpdate):
        """处理订单待更新"""
        self.lifecycle_manager.update_order_state(event)

    def on_order_pending_cancel(self, event: OrderPendingCancel):
        """处理订单待取消"""
        self.lifecycle_manager.update_order_state(event)

    def on_order_filled(self, event: OrderFilled):
        """处理订单成交"""
        self.lifecycle_manager.update_order_state(event)
        self._handle_fill(event)

    def on_order_partially_filled(self, event: OrderPartiallyFilled):
        """处理部分成交"""
        self.lifecycle_manager.update_order_state(event)

    def _handle_rejection(self, event: OrderRejected):
        """处理订单拒绝"""
        self.log.error(
            f"订单被拒绝: {event.order_id}, "
            f"原因: {event.reason}, "
            f"策略: {event.strategy_id}"
        )

        # 这里可以实现重试逻辑
        self._maybe_retry_order(event.order)

    def _handle_fill(self, event: OrderFilled):
        """处理订单成交"""
        self.log.info(
            f"订单成交: {event.order_id}, "
            f"数量: {event.last_qty}, "
            f"价格: {event.last_px}, "
            f"手续费: {event.commission}"
        )

        # 更新策略状态
        self._update_strategy_after_fill(event)

    def _update_strategy_after_fill(self, event: OrderFilled):
        """更新策略状态（子类实现）"""
        pass

    def _maybe_retry_order(self, rejected_order):
        """可能重试被拒绝的订单（子类实现）"""
        pass

    def get_order_stats(self):
        """获取订单统计"""
        return self.lifecycle_manager.get_order_stats()

8.3 订单组合策略

8.3.1 OCO（One Cancels Other）

"""
OCO订单策略
"""

from nautilus_trader.model.orders import OrderList
from nautilus_trader.model.enums import OrderType


class OCOOrderStrategy(Strategy):
    """OCO订单策略"""

    def __init__(self):
        """初始化策略"""
        super().__init__()

        self.oco_order_lists = {}  # oco_id -> order_list
        self.active_oco_sets = {}  # oco_id -> orders

    def create_oco_orders(
        self,
        take_profit_price: Price,
        stop_loss_price: Price,
        quantity: Decimal,
    ) -> OrderList:
        """创建OCO订单组"""

        # 止盈限价单
        take_profit_order = LimitOrder(
            trader_id=self.trader_id,
            strategy_id=self.id,
            instrument_id=self.bar_type.instrument_id,
            order_side=OrderSide.SELL,
            quantity=Quantity(quantity),
            price=take_profit_price,
            time_in_force=TimeInForce.GTC,
            reduce_only=True,
        )

        # 止损市价单
        stop_loss_order = StopMarketOrder(
            trader_id=self.trader_id,
            strategy_id=self.id,
            instrument_id=self.bar_type.instrument_id,
            order_side=OrderSide.SELL,
            quantity=Quantity(quantity),
            trigger_price=stop_loss_price,
            time_in_force=TimeInForce.GTC,
            reduce_only=True,
        )

        # 创建OCO订单列表
        oco_orders = OrderList(
            orders=[take_profit_order, stop_loss_order],
            oco=True,  # 一成交另一取消
        )

        self.oco_order_lists[oco_orders.id] = oco_orders
        self.active_oco_sets[oco_orders.id] = {
            'created_at': datetime.utcnow(),
            'take_profit_id': take_profit_order.client_order_id,
            'stop_loss_id': stop_loss_order.client_order_id,
            'filled': False,
            'canceled': False,
        }

        self.log.info(
            f"创建OCO订单组 - ID: {oco_orders.id}, "
            f"止盈: {take_profit_price}, "
            f"止损: {stop_loss_price}"
        )

        return oco_orders

    def submit_oco_orders(self, oco_orders: OrderList):
        """提交OCO订单"""
        self.submit_order_list(oco_orders)

    def on_order_filled(self, event: OrderFilled):
        """处理订单成交"""
        # 检查是否是OCO订单
        for oco_id, oco_info in self.active_oco_sets.items():
            if event.order_id in [oco_info['take_profit_id'], oco_info['stop_loss_id']]:
                # 标记OCO组已成交
                oco_info['filled'] = True
                oco_info['filled_price'] = event.last_px
                oco_info['filled_time'] = datetime.fromtimestamp(event.ts_event / 1e9)

                # 另一个订单应该被自动取消
                self._cancel_oco_partner(oco_id, event.order_id)

                # 记录OCO执行结果
                self._log_oco_result(oco_id, event.order_id, event)

                break

    def on_order_canceled(self, event: OrderCanceled):
        """处理订单取消"""
        # 检查是否是OCO订单的伙伴被取消
        for oco_id, oco_info in self.active_oco_sets.items():
            if event.order_id in [oco_info['take_profit_id'], oco_info['stop_loss_id']]:
                if not oco_info['filled']:
                    oco_info['canceled'] = True
                    oco_info['canceled_reason'] = 'PARTNER_CANCELED'
                    self._log_oco_cancellation(oco_id, event)
                break

    def _cancel_oco_partner(self, oco_id: str, filled_order_id: str):
        """取消OCO伙伴订单"""
        oco_info = self.active_oco_sets[oco_id]

        # 确定要取消的订单
        if filled_order_id == oco_info['take_profit_id']:
            partner_id = oco_info['stop_loss_id']
            partner_type = '止损单'
        else:
            partner_id = oco_info['take_profit_id']
            partner_type = '止盈单'

        # 取消伙伴订单
        if partner_id:
            self.cancel_order(partner_id)
            self.log.info(f"OCO执行 - 自动取消{partner_type}: {partner_id}")

    def _log_oco_result(self, oco_id: str, filled_order_id: str, fill_event):
        """记录OCO执行结果"""
        oco_info = self.active_oco_sets[oco_id]

        filled_type = '止盈' if filled_order_id == oco_info['take_profit_id'] else '止损'
        duration = oco_info['filled_time'] - oco_info['created_at']

        self.log.info(
            f"OCO订单组{oco_id}执行完成 - "
            f"{filled_type}成交: {fill_event.last_px}, "
            f"执行时间: {duration.total_seconds():.1f}秒"
        )

    def _log_oco_cancellation(self, oco_id: str, cancel_event):
        """记录OCO取消"""
        oco_info = self.active_oco_sets[oco_id]

        partner_order_id = (oco_info['take_profit_id'] if cancel_event.order_id == oco_info['stop_loss_id']
                        else oco_info['stop_loss_id'])

        partner_type = '止损单' if partner_order_id == oco_info['stop_loss_id'] else '止盈单'

        self.log.info(
            f"OCO订单组{oco_id}部分执行 - "
            f"取消{partner_type}: {partner_order_id}"
        )

    def cleanup_completed_oco_sets(self):
        """清理已完成的OCO组"""
        completed_oco = []

        for oco_id, oco_info in self.active_oco_sets.items():
            if oco_info['filled']:
                completed_oco.append(oco_id)
                continue

            # 检查所有订单是否都已完成
            all_completed = True
            for order_id in [oco_info['take_profit_id'], oco_info['stop_loss_id']]:
                state = self.cache.order(order_id)
                if state and state.status not in ['FILLED', 'CANCELED']:
                    all_completed = False
                    break

            if all_completed:
                completed_oco.append(oco_id)

        # 清理
        for oco_id in completed_oco:
            del self.active_oco_sets[oco_id]
            if oco_id in self.oco_order_lists:
                del self.oco_order_lists[oco_id]

        if completed_oco:
            self.log.info(f"清理 {len(completed_oco)} 个OCO订单组")


# 使用OCO策略的示例
class OCOTradingStrategy(Strategy):
    """使用OCO的交易策略"""

    def __init__(self):
        """初始化策略"""
        super().__init__()
        self.oco_manager = OCOOrderStrategy()
        self.current_position = None

    def on_order_filled(self, event: OrderFilled):
        """处理订单成交"""
        self.oco_manager.on_order_filled(event)

        # 检查是否是开仓
        if event.order_side == OrderSide.BUY:
            self._handle_position_open(event)
        # OCO成交应该是平仓
        else:
            self._handle_position_close(event)

    def _handle_position_open(self, event: OrderFilled):
        """处理开仓"""
        self.current_position = {
            'entry_price': event.last_px,
            'quantity': event.last_qty,
            'entry_time': datetime.fromtimestamp(event.ts_event / 1e9),
        }

        # 计算止盈止损价格
        take_profit_price = self._calculate_take_profit(event.last_px)
        stop_loss_price = self._calculate_stop_loss(event.last_px)

        # 创建OCO订单
        oco_orders = self.oco_manager.create_oco_orders(
            take_profit_price,
            stop_loss_price,
            event.last_qty,
        )

        # 提交OCO订单
        self.oco_manager.submit_oco_orders(oco_orders)

        self.log.info(
            f"开仓成功 - 价格: {event.last_px}, "
            f"设置OCO止盈: {take_profit_price}, "
            f"止损: {stop_loss_price}"
        )

    def _calculate_take_profit(self, entry_price: Decimal) -> Price:
        """计算止盈价格"""
        risk_reward_ratio = 2.0  # 1:2 的风险收益比
        risk_amount = entry_price * Decimal("0.02")  # 2% 风险
        take_profit_amount = risk_amount * risk_reward_ratio
        take_profit_price = entry_price + take_profit_price

        return Price(take_profit_price)

    def _calculate_stop_loss(self, entry_price: Decimal) -> Price:
        """计算止损价格"""
        stop_distance = entry_price * Decimal("0.02")  # 2% 止损
        stop_loss_price = entry_price - stop_distance

        return Price(stop_loss_price)

    def _handle_position_close(self, event: OrderFilled):
        """处理平仓"""
        if self.current_position:
            duration = datetime.fromtimestamp(event.ts_event / 1e9) - self.current_position['entry_time']
            pnl = (event.last_px - self.current_position['entry_price']) * self.current_position['quantity']

            self.log.info(
                f"平仓完成 - 入场价: {self.current_position['entry_price']}, "
                f"出场价: {event.last_px}, "
                f"盈亏: {pnl}, "
                f"持仓时间: {duration.total_seconds():.1f}秒"
            )

            self.current_position = None

            # 清理OCO订单
            self.oco_manager.cleanup_completed_oco_sets()

8.3.2 OTO（One Triggers Other）

"""
OTO订单策略
"""

from nautilus_trader.model.orders import OrderList
from nautilus_trader.model.enums import OrderType


class OTOOrderStrategy(Strategy):
    """OTO订单策略"""

    def __init__(self):
        """初始化策略"""
        super().__init__()

        self.oto_order_lists = {}  # oto_id -> order_list
        self.active_oto_sets = {}  # oto_id -> orders

    def create_oto_orders(
        self,
        primary_order,  # 主要订单（限价单）
        secondary_order,  # 触发订单（止损单）
    ) -> OrderList:
        """创建OTO订单组"""

        # 创建OTO订单列表
        oto_orders = OrderList(
            orders=[primary_order, secondary_order],
            oto=True,  # 一触发另一个
        )

        self.oto_order_lists[oto_orders.id] = oto_orders
        self.active_oto_sets[oto_orders.id] = {
            'created_at': datetime.utcnow(),
            'primary_id': primary_order.client_order_id,
            'secondary_id': secondary_order.client_order_id,
            'triggered': False,
            'canceled': False,
        }

        self.log.info(
            f"创建OTO订单组 - ID: {oto_orders.id}, "
            f"主要订单: {primary_order}, "
            f"触发订单: {secondary_order}"
        )

        return oto_orders

    def submit_oto_orders(self, oto_orders: OrderList):
        """提交OTO订单"""
        self.submit_order_list(oto_orders)

    def on_order_filled(self, event: OrderFilled):
        """处理订单成交"""
        # 检查是否是OTO订单的触发条件
        for oto_id, oto_info in self.active_oto_sets.items():
            if event.order_id == oto_info['primary_id']:
                # 主要订单成交，触发次要订单
                self._trigger_secondary_order(oto_id, event)
                break

    def on_order_canceled(self, event: OrderCanceled):
        """处理订单取消"""
        # 检查是否是OTO订单
        for oto_id, oto_info in self.active_oto_sets.items():
            if event.order_id == oto_info['primary_id']:
                # 主要订单取消，自动取消次要订单
                self._cancel_secondary_order(oto_id, event)
                break

    def _trigger_secondary_order(self, oto_id: str, trigger_event):
        """触发次要订单"""
        oto_info = self.active_oto_sets[oto_id]

        if not oto_info['triggered']:
            oto_info['triggered'] = True
            oto_info['triggered_at'] = datetime.fromtimestamp(trigger_event.ts_event / 1e9)

            # 获取次要订单并提交
            secondary_order = self.cache.order(oto_info['secondary_id'])
            if secondary_order and secondary_order.status == OrderStatus.INITIALIZED:
                self.submit_order(secondary_order)
                self.log.info(
                    f"OTO触发 - 提交次要订单: {secondary_order}, "
                    f"触发价格: {trigger_event.last_px}"
                )

    def _cancel_secondary_order(self, oto_id: str, cancel_event):
        """取消次要订单"""
        oto_info = self.active_oto_sets[oto_id]

        if not oto_info['triggered']:
            # 取消未触发的次要订单
            secondary_order_id = oto_info['secondary_id']
            secondary_order = self.cache.order(secondary_order_id)

            if secondary_order and secondary_order.status in [OrderStatus.INITIALIZED, OrderStatus.SUBMITTED]:
                self.cancel_order(secondary_order_id)
                self.log.warning(
                    f"OTO取消 - 主要订单取消，自动取消次要订单: {secondary_order}"
                )

            oto_info['canceled'] = True
            oto_info['canceled_at'] = datetime.utcnow()
            oto_info['cancel_reason'] = 'PRIMARY_CANCELLED'


# 使用OTO策略的示例
class OTOMomentumStrategy(Strategy):
    """使用OTO的动量策略"""

    def __init__(self):
        """初始化策略"""
        super().__init__()
        self.oto_manager = OTOOrderStrategy()

        # 指标
        self.rsi = RSI(period=14)
        self.ema = EMA(period=20)

        # 状态
        self.position = None
        self.entry_oto_id = None

    def on_bar(self, bar: Bar):
        """处理K线数据"""
        # 更新指标
        self.rsi.update_raw(float(bar.close))
        self.ema.update_raw(float(bar.close))

        if not (self.rsi.initialized and self.ema.initialized):
            return

        # 检查交易信号
        if not self.position:
            self._check_entry_signal(bar)
        else:
            self._check_exit_signal(bar)

    def _check_entry_signal(self, bar: Bar):
        """检查入场信号"""
        current_price = bar.close
        ema_value = self.ema.value
        rsi_value = self.rsi.value

        # 买入条件：价格 > EMA 且 RSI > 50
        if (current_price > ema_value and rsi_value > 50 and
            not self.entry_oto_id):

            # 设置买入限价单（低于当前价格）
            entry_price = current_price * Decimal("0.995")

            primary_order = LimitOrder(
                trader_id=self.trader_id,
                strategy_id=self.id,
                instrument_id=self.bar_type.instrument_id,
                order_side=OrderSide.BUY,
                quantity=Quantity(Decimal("0.01")),
                price=Price(entry_price),
                time_in_force=TimeInForce.GTC,
            )

            # 设置止损单作为触发订单
            stop_loss_price = entry_price * Decimal("0.98")

            secondary_order = StopMarketOrder(
                trader_id=self.trader_id,
                strategy_id=self.id,
                instrument_id=self.bar_type.instrument_id,
                order_side=OrderSide.SELL,
                quantity=Quantity(Decimal("0.01")),
                trigger_price=Price(stop_loss_price),
                time_in_force=TimeInForce.GTC,
                reduce_only=True,
            )

            # 创建OTO订单
            oto_orders = self.oto_manager.create_oto_orders(
                primary_order,
                secondary_order,
            )

            # 提交OTO
            self.oto_manager.submit_oto_orders(oto_orders)
            self.entry_oto_id = oto_orders.id

            self.log.info(
                f"提交OTO - 买单: {entry_price}, "
                f"止损触发: {stop_loss_price}"
            )

    def _check_exit_signal(self, bar: Bar):
        """检查出场信号"""
        current_price = bar.close
        entry_price = self.position['entry_price']

        # 简单的止盈策略
        if current_price > entry_price * Decimal("1.03"):
            # 市价卖出
            exit_order = MarketOrder(
                trader_id=self.trader_id,
                strategy_id=self.id,
                instrument_id=self.bar_type.instrument_id,
                order_side=OrderSide.SELL,
                quantity=self.position['quantity'],
            )
            self.submit_order(exit_order)
            self.log.info(f"止盈出场 - 价格: {current_price}")

    def on_order_filled(self, event: OrderFilled):
        """处理订单成交"""
        self.oto_manager.on_order_filled(event)

        if event.order_side == OrderSide.BUY and not self.position:
            # 开仓
            self.position = {
                'entry_price': event.last_px,
                'quantity': event.last_qty,
                'entry_time': datetime.fromtimestamp(event.ts_event / 1e9),
            }
            self.log.info(f"开仓 - 价格: {event.last_px}")

        elif event.order_side == OrderSide.SELL and self.position:
            # 平仓
            pnl = (event.last_px - self.position['entry_price']) * self.position['quantity']
            self.log.info(
                f"平仓 - 盈亏: {pnl}, "
                f"入场价: {self.position['entry_price']}, "
                f"出场价: {event.last_px}"
            )
            self.position = None
            self.entry_oto_id = None

    def on_order_rejected(self, event: OrderRejected):
        """处理订单拒绝"""
        self.oto_manager.on_order_rejected(event)

        # 如果主要订单被拒绝，重试
        if self.entry_oto_id:
            for oto_id, oto_info in self.oto_manager.active_oto_sets.items():
                if (oto_id == self.entry_oto_id and
                    event.order_id == oto_info['primary_id']):
                    # 清理OTO并重试
                    del self.oto_manager.active_oto_sets[oto_id]
                    self.entry_oto_id = None
                    break

8.4 订单执行算法

8.4.1 TWAP（时间加权平均价格）

"""
TWAP执行算法
"""

from decimal import Decimal
from typing import List, Dict
from datetime import datetime, timedelta
import numpy as np


class TWAPExecutor:
    """TWAP执行器"""

    def __init__(
        self,
        trader_id,
        strategy_id,
        instrument_id,
        total_quantity: Decimal,
        duration_seconds: int,
        num_slices: int = 10,
    ):
        """
        初始化TWAP执行器

        Parameters
        ----------
        trader_id : TraderId
            交易者ID
        strategy_id : StrategyId
            策略ID
        instrument_id : InstrumentId
            交易工具ID
        total_quantity : Decimal
            总执行数量
        duration_seconds : int
            执行持续时间（秒）
        num_slices : int
            分片数量
        """
        self.trader_id = trader_id
        self.strategy_id = strategy_id
        self.instrument_id = instrument_id

        self.total_quantity = total_quantity
        self.remaining_quantity = total_quantity

        self.duration = duration_seconds
        self.num_slices = num_slices
        self.slice_interval = duration_seconds / num_slices

        self.slice_quantities = self._calculate_slice_quantities()

        self.orders = []  # 已提交的订单
        self.filled_orders = []  # 已成交的订单
        self.start_time = None
        self.end_time = None

        self.is_active = False
        self.is_completed = False

    def _calculate_slice_quantities(self) -> List[Decimal]:
        """计算分片数量"""
        # 均匀分配
        base_quantity = self.total_quantity / self.num_slices

        # 添加随机扰动
        random_factors = np.random.uniform(0.8, 1.2, self.num_slices)
        random_factors /= np.mean(random_factors)  # 归一化

        quantities = [base_quantity * Decimal(str(f)) for f in random_factors]
        quantities = [round(q, 6) for q in quantities]

        # 确保总和等于总量
        total = sum(quantities)
        if total != self.total_quantity:
            # 调整最后一个分片
            quantities[-1] += (self.total_quantity - total)

        return quantities

    def start_execution(self, start_time_ns: int):
        """开始执行"""
        self.start_time = datetime.fromtimestamp(start_time_ns / 1e9)
        self.end_time = self.start_time + timedelta(seconds=self.duration)
        self.is_active = True

        # 创建并提交所有订单
        self._create_and_submit_orders()

    def _create_and_submit_orders(self):
        """创建并提交所有订单"""
        current_time = self.start_time
        remaining = self.remaining_quantity

        for i, (slice_qty, slice_num) in enumerate(
            zip(self.slice_quantities, range(self.num_slices))
        ):
            if slice_qty <= 0:
                continue

            # 计算订单时间
            order_time = current_time + timedelta(
                seconds=(slice_num + 0.5) * self.slice_interval
            )

            # 使用限价单，价格基于当前市场价
            # 这里应该获取当前价格
            current_price = self._get_current_price()

            # 添加随机价格扰动
            price_adjustment = np.random.uniform(-0.0005, 0.0005)  # ±0.05%
            limit_price = current_price * (1 + price_adjustment)

            # 创建限价单
            order = LimitOrder(
                trader_id=self.trader_id,
                strategy_id=self.strategy_id,
                instrument_id=self.instrument_id,
                order_side=OrderSide.BUY,  # 可以根据需要设置
                quantity=Quantity(slice_qty),
                price=Price(limit_price),
                time_in_force=TimeInForce.GTC,
                expire_time=int(order_time.timestamp() * 1e9),
                post_only=True,  # 只做maker，避免手续费
            )

            # 这里应该提交订单
            self.orders.append({
                'order': order,
                'slice_number': slice_num,
                'quantity': slice_qty,
                'scheduled_time': order_time,
                'status': 'SCHEDULED',
            })

    def _get_current_price(self) -> Decimal:
        """获取当前价格（示例实现）"""
        # 这里应该从缓存或数据引擎获取
        # 返回模拟值
        return Decimal("50000.00")

    def update_execution(self, current_time_ns: int):
        """更新执行状态"""
        if not self.is_active or self.is_completed:
            return

        current_time = datetime.fromtimestamp(current_time_ns / 1e9)

        # 检查是否超时
        if current_time > self.end_time:
            self._complete_execution("TIMEOUT")
            return

        # 检查是否已完成
        if self.remaining_quantity <= 0:
            self._complete_execution("COMPLETED")
            return

        # 这里可以添加动态调整逻辑
        # 例如基于市场条件调整剩余订单

    def on_order_filled(self, event):
        """处理订单成交"""
        order_id = event.order_id

        # 找到对应的订单信息
        for order_info in self.orders:
            if order_info['order'].client_order_id == order_id:
                order_info['status'] = 'FILLED'
                order_info['fill_time'] = datetime.fromtimestamp(event.ts_event / 1e9)
                order_info['fill_price'] = event.last_px
                order_info['fill_qty'] = event.last_qty

                self.filled_orders.append(order_info)
                self.remaining_quantity -= event.last_qty

                break

        # 检查是否需要取消剩余订单
        if self.remaining_quantity <= 0:
            self._cancel_remaining_orders()

    def _cancel_remaining_orders(self):
        """取消剩余订单"""
        for order_info in self.orders:
            if order_info['status'] == 'SCHEDULED':
                # 这里应该取消订单
                order_info['status'] = 'CANCELED'

    def _complete_execution(self, reason: str):
        """完成执行"""
        self.is_active = False
        self.is_completed = True
        self.completion_time = datetime.utcnow()
        self.completion_reason = reason

        # 取消所有未成交的订单
        self._cancel_remaining_orders()

    def get_execution_stats(self) -> dict:
        """获取执行统计"""
        total_qty = float(self.total_quantity)
        filled_qty = sum(float(o['fill_qty']) for o in self.filled_orders)

        if filled_qty > 0:
            avg_price = sum(
                float(o['fill_price']) * float(o['fill_qty'])
                for o in self.filled_orders
            ) / filled_qty
        else:
            avg_price = 0

        filled_rate = filled_qty / total_qty if total_qty > 0 else 0

        return {
            'total_quantity': total_qty,
            'filled_quantity': filled_qty,
            'remaining_quantity': float(self.remaining_quantity),
            'filled_rate': filled_rate,
            'average_fill_price': avg_price,
            'total_orders': len(self.orders),
            'filled_orders': len(self.filled_orders),
            'is_completed': self.is_completed,
            'completion_reason': getattr(self, 'completion_reason', None),
        }


# 使用TWAP的策略
class TWAPStrategy(Strategy):
    """使用TWAP执行算法的策略"""

    def __init__(self):
        """初始化策略"""
        super().__init__()

        self.twap_executors = {}  # execution_id -> executor
        self.next_execution_id = 1

    def create_twap_execution(
        self,
        quantity: Decimal,
        duration_minutes: int,
        num_slices: int = 20,
    ) -> str:
        """创建TWAP执行"""

        execution_id = f"TWAP_{self.next_execution_id}"
        self.next_execution_id += 1

        executor = TWAPExecutor(
            trader_id=self.trader_id,
            strategy_id=self.id,
            instrument_id=self.bar_type.instrument_id,
            total_quantity=quantity,
            duration_seconds=duration_minutes * 60,
            num_slices=num_slices,
        )

        self.twap_executors[execution_id] = executor
        return execution_id

    def start_twap_execution(self, execution_id: str):
        """开始TWAP执行"""
        if execution_id not in self.twap_executors:
            self.log.error(f"TWAP执行器不存在: {execution_id}")
            return

        executor = self.twap_executors[execution_id]
        executor.start_execution(self.clock.timestamp_ns())

        # 提交所有订单
        for order_info in executor.orders:
            self.submit_order(order_info['order'])

        self.log.info(f"开始TWAP执行: {execution_id}")

    def update_twap_executions(self):
        """更新TWAP执行状态"""
        current_time = self.clock.timestamp_ns()

        for executor in self.twap_executors.values():
            executor.update_execution(current_time)

    def on_order_filled(self, event):
        """处理订单成交"""
        # 检查是否是TWAP订单
        for executor in self.twap_executors.values():
            executor.on_order_filled(event)

    def get_twap_stats(self, execution_id: str) -> dict:
        """获取TWAP统计"""
        if execution_id not in self.twap_executors:
            return {}

        return self.twap_executors[execution_id].get_execution_stats()

    def on_bar(self, bar: Bar):
        """处理K线数据"""
        # 更新TWAP执行
        self.update_twap_executions()

        # 简单的示例：每1小时创建一个新的TWAP执行
        if self.count_bars() % 60 == 0:  # 假设1小时K线
            # 创建0.1 BTC的TWAP执行，持续30分钟
            execution_id = self.create_twap_execution(
                quantity=Decimal("0.1"),
                duration_minutes=30,
                num_slices=30,
            )

            # 立即开始执行
            self.start_twap_execution(execution_id)

    def on_stop(self):
        """策略停止"""
        # 完成所有TWAP执行
        for executor in self.twap_executors.values():
            if executor.is_active:
                executor._complete_execution("STRATEGY_STOP")

        # 输出统计
        self._log_twap_stats()

    def _log_twap_stats(self):
        """记录TWAP统计"""
        self.log.info("TWAP执行统计:")

        for execution_id, executor in self.twap_executors.items():
            stats = executor.get_execution_stats()
            if stats:
                self.log.info(
                    f"执行器 {execution_id}: "
                    f"执行率: {stats['filled_rate']:.1%}, "
                    f"平均价格: {stats['average_fill_price']:.2f}"
                )

8.5 下一步

在本章中，我们学习了：

NautilusTrader 订单系统的核心概念
各种订单类型的使用方法
订单生命周期管理
高级订单策略（OCO、OTO）
执行算法（TWAP）

在下一章中，我们将学习：

投资组合管理
风险控制机制
资金管理策略
多策略协调
绩效分析

DEV Community: Henry Lin

RD-AGENT 第六讲 · 一次预注册实验的完整生命周期：两份判决书

第六讲 · 一次预注册实验的完整生命周期：两份判决书

1. 为什么选这两个案例

2. 案例一：业绩预告事件窗 —— REJECTED

2.1 假设

2.2 预注册要点

2.3 判决

2.4 判决的执行——最见制度成色的部分

3. 案例二：指数调样流 —— INCONCLUSIVE

3.1 假设

3.2 流程与装配验证

3.3 判决

3.4 INCONCLUSIVE 的正确处理（比 REJECTED 难）

4. 三种判决的决策树（本章收束）

5. 第二章总结

RD-AGENT 第五讲 · 三候选 DSR 全 FAIL 复盘：超额从哪来，又到哪去了

第五讲 · 三候选 DSR 全 FAIL 复盘：超额从哪来，又到哪去了

1. 案例背景

2. 第一刀：DSR——你试了多少次，就要还多少债

3. 第二刀：归因分解——超额是 alpha 还是风格 beta

4. 第三刀：换手与成本——毛收益买不回净收益

5. 最深的一刀：基准伪影（预告第八讲）

6. 复盘：四刀的层级结构

7. 军心稳定问题：全 FAIL 之后怎么办

8. 本讲小结

RD-AGENT 第四讲 · 为什么需要独立验证：预注册与验证管线

第四讲 · 为什么需要独立验证：预注册与验证管线

1. 从一个问题开始

2. 多重检验：为什么"边看边调"必然产出假阳性

3. 验证管线的四个构件

构件一：独立数据复算

构件二：预注册门槛

构件三：样本外一次性判定（OOS 一次性）

构件四：留痕与可复现

4. 一次性判定的三份真实判决书

5. 本讲小结

RD-AGENT 第三讲 · LLM 写的因子长什么样：AI 的研究品味

第三讲 · LLM 写的因子长什么样：AI 的研究品味

1. 开场

2. 案例一：像样的——但要看清"像样"的来源

2.1 一个必须讲的反面注脚：carry 因子的 +18% 是假的

3. 案例二：平庸的——伪逻辑的识别

4. 案例三：翻车的——以及翻车之后发生了什么

5. 本章总结

RD-AGENT 第二讲 · 看懂排行榜：IC、超额收益与因子版本

第二讲 · 看懂排行榜：IC、超额收益与因子版本

1. 一张真实的排行榜

2. 四个指标：直觉与陷阱

IC（信息系数）

Rank IC

ICIR

超额收益（excess return）

3. 因子的"组合"与"版本"：为什么一个因子有十几条记录

4. 自报成绩的三个"没告诉你"

5. 本讲小结

RD-AGENT 第一讲 · AI 因子工厂是怎么运转的

第一讲 · AI 因子工厂是怎么运转的

1. 开场：一台真实跑过的机器

2. 循环的五步

3. 一次循环执行记录长什么样

4. 工厂的日常：一半时间在跟基础设施搏斗

5. 人站在哪里

6. 本讲小结

量化开源项目v6.0已经更新

henrylin99 / quantitative_analysis

量化分析

多因子选股系统

项目定位

能力现状（已实现 / 部分实现 / 未实现/未开放）

功能概览

🤖 AI 智能工作台

量化开源软件v3.0发布

DeepSeek Harness (`dsh`) Agent 开发教程

DeepSeek Harness (dsh) Agent 开发教程

目录

1. Agent 在架构中的位置

2. 五个核心概念

2.1 Agent：活 agent 的编程面

2.2 ctx.agents：注册表与创建入口

DeepSeek Harness (`dsh`) Agent 开发教程

2.1 `Agent`：活 agent 的编程面

2.2 `ctx.agents`：注册表与创建入口

2.4 `SessionId` 与品牌化 ID

4.2 用 `agents` 配置项预先创建 agent

7.1 `agent/*` 事件速查

DeepSeek Harness (`dsh`) 使用教程