From 58e1e3e29179086917859554596be7e71f43becd Mon Sep 17 00:00:00 2001 From: yanweidong Date: Sat, 5 Sep 2026 13:53:26 +0800 Subject: [PATCH] fix bug --- docs/ipo-trend-audit.md | 215 +++++++++++------ docs/trend-state.md | 37 +++ py-client/strategy/trend/boot.py | 9 +- py-client/strategy/trend/open.py | 8 +- py-client/strategy/trend/order.py | 47 +--- py-client/strategy/trend/positions.py | 10 +- py-client/strategy/trend/state.py | 334 ++++++++++++-------------- 7 files changed, 343 insertions(+), 317 deletions(-) create mode 100644 docs/trend-state.md diff --git a/docs/ipo-trend-audit.md b/docs/ipo-trend-audit.md index 5817c68..028012e 100644 --- a/docs/ipo-trend-audit.md +++ b/docs/ipo-trend-audit.md @@ -1,109 +1,164 @@ # IPO / Trend 当前代码审计 -日期：2026-09-05。范围：`py-client/strategy/{ipo,trend}`，以及直接依赖的 SDK、公共库、配置与 `api/qmt_rest_new.py` 下单/撤单处理器。以下路径除特别注明外均相对于 `py-client/`。 +> 后续实现更新：已按要求精简为 JSON v3，订单终态后一次记账，见 [trend-state.md](trend-state.md)。T12 的持仓缺席自动删除已移除；T13 的提交前写盘失败已增加占位回滚并验证。下方对应复现保留为 v2 历史证据。其余未修复项仍需单独处理；v3 不再保存子单历史及成交增量，旧版本迁移与清仓后重新开仓的操作边界见新说明。 -本报告取代上一轮 Trend 审计的当前结论。仅清理测试文件、更新文档；未实施策略修复。采用静态阅读与隔离的内存断言，不连接柜台、不提交真实订单。风险等级按状态丢失、重复委托、漏单等后果划分；未验证真实 QMT 的返回值与回报时序。 +审计日期：2026-09-05；本轮重新读取当前工作树，重点复核 JSON v2 状态机。范围为 `py-client/strategy/{ipo,trend}` 及直接依赖，不包含 ZT 策略。以下路径相对于 `py-client/`，服务端路径从仓库根目录起算。 -## IPO：高风险 +本轮只更新审计文档，未修改策略源码。已清理的 tests/*.py 不重建；采用临时目录和 mock 进行隔离复现，没有调用真实柜台。本文取代上一轮联合审计的当前结论；状态格式说明见 [trend-state.md](trend-state.md)。 -### I1. 未确认受理就写入永久申购标记 （忽略） +## 结论 -位置：`strategy/ipo/boot.py:53`、`libs/lockfile.py:14`；服务端 `api/qmt_rest_new.py:246`。 +新的 pending 与业务账本分离、补仓首次成交计数、预开仓数量落盘、完整订单快照对账均已接入。但尚不能据此认定状态恢复与异常生命周期完整：空持仓快照仍可清空历史补仓记录；确定未提交/未受理的订单可能永久占用 pending；撤单异常仍可阻断整轮状态对账。 -`client.passorder()` 返回值被忽略，随后无条件写锁、记录申购日志。服务端只要底层调用未抛异常就返回 success，并直接把返回值字符串化，未验证引用是否有效。因此本地标记只能证明调用返回，不能证明委托有效。遇到失败业务响应或无效引用，后续轮次仍因锁文件存在而跳过，导致漏申购。 +优先处理 T12、T13、T14、T15。下文沿用原编号，新发现从 T12 起编号，不把已解决的问题重复列为当前缺陷。 -建议：依据明确的柜台受理契约验证结果；无法确认的结果保留待核查记录，按委托/成交查询确认后再标记完成。 +## 新发现：Trend -### I2. 检查锁、下单、写锁不是原子操作，且没有券商对账（忽略） +### T12【高】一次空持仓快照会清除已完成账本，恢复后补仓次数归零 + +位置：`strategy/trend/state.py:212`、`state.py:124`；入口 `boot.py:182`。 + +证据：对所有“曾出现于持仓、当前不在持仓、无 pending”的证券立即删除 items 与 seen_positions。持仓下次恢复时按首次接管重新建立底仓，added_num 从零开始。SDK 的组合解析还会将缺失 positions 字段视为空字典。 + +复现 R1：已有底仓并完成一次补仓 → reconcile([], []) → 原记录被删除 → 持仓恢复 → added_num 从 1 变成 0。 + +影响：历史补仓量价丢失，亏损补仓档位可能重新使用。这与已修复的“空订单快照丢 pending”是不同路径。当前文档已声明依赖持仓快照完整，但代码没有验证此条件。 + +建议：持仓缺席进入核查流程，不立即删除账本；结合卖出成交、明确清仓或可靠的持仓二次查询确认后结束证券生命周期。缺失字段不能自动视为有效空账户。 + +### T13【高】提交前写盘失败，会留下实际上未发送的 pending + +位置：`strategy/trend/state.py:79`、`order.py:114`。 + +证据：begin 先修改 items、pending、索引和 dirty，再调用 save。save 抛错后没有回滚；place 因异常没有调用柜台。后续 reconcile 的 save 可以把这个未发送记录写入磁盘。 + +复现 R2：模拟 Path.write_text 抛出磁盘错误；柜台调用次数为 0，但 busy(A) 为真；磁盘恢复后执行 reconcile 并重载文件，busy(A) 仍为真。 + +影响：该证券持续被阻止买入；后续没有真实订单可供终态对账，180 秒日志也不会解除锁定。 + +建议：将提交前登记视为内存/磁盘事务，明确写盘失败时回滚本次占位，或记录“确定未发送”并走安全撤销流程；与请求发出后结果不确定的情形区分。 + +### T14【高】服务端明确 HTTP 400 拒绝也被永久保留为待确认 + +位置：`strategy/trend/order.py:127`、`sdk/client.py:67`、服务端 `api/qmt_rest_new.py:235`。 + +证据：服务端参数解析失败时在实际 passorder 调用前抛 HTTP 400；SDK 转为 APIError；place 仅记录并返回 False，未调用 state.reject。当前 reject 分支只识别响应字典中的 failed/rejected，而本服务端下单处理器正常只返回 success，失败主要通过 HTTP 错误表达。 + +复现 R3：mock APIError(400)，清除 SimpleCache 并重新加载 State 后，证券仍 busy。 + +影响：确定未受理的请求无法自动恢复；纠正参数后仍不能下单。 + +建议：针对已确认的服务端契约区分“明确未受理”和“不确定”。明确的参数拒绝可结束 pending；网络超时、502 等不能一概按失败释放。 + +### T15【高】撤单异常仍会阻断完整状态对账及整轮持仓管理 + +位置：`strategy/trend/order.py:86`、`boot.py:132`、`boot.py:182`。 + +证据：refresh 在遍历订单中直接调用 cancel_by_id；一次异常会使 refresh 退出，RunOnce 在组合刷新异常分支直接 return，已经拿到的其他订单成交快照也不会进入 reconcile。启动阶段同类异常会导致启动失败。 + +复现 R5：快照包含超时委托，撤单接口抛错；state.reconcile 调用次数为 0。 + +影响：单个撤单故障可以持续阻断全部证券的成交更新、止盈和补仓。T6 的逐持仓隔离在这个阶段尚未执行。 + +建议：先完成快照采集及状态对账；撤单采用逐订单异常边界，失败保持 busy，并记录业务失败结果，不阻断其他订单和证券。 + +## 新状态机的条件性风险与恢复边界 + +### T16【中，条件性】结束 pending 后不能处理迟到的成交金额修正 + +位置：`strategy/trend/state.py:107`、`state.py:197`。 + +pending 结束即删除子单数据和已记账基准，后续只遍历 pending，不再处理该订单。复现 R4：100 股终态成交金额 1000 元完成后，再收到金额 1050 元的同订单快照，base_cost 仍为 10 元。 + +是否实际触发取决于柜台终态金额是否可能修正，本轮没有实盘证据。若上游保证终态价格最终不变，可将其作为明确契约；否则应短期保留已完成订单的记账基准并支持差额修正。 + +### 迁移与核查：已知限制，不认定为新回归 + +位置：`strategy/trend/state.py:233`。 + +- 旧 ING 的 expected_qty 迁移为 0，完成条件要求大于 0，因此必须人工核实后补齐；当前没有专门的核查/解除接口。 +- 旧文件可能只保存最后一次补仓量价，新累计 added_amount 不能凭空恢复全部历史；迁移有备份和告警，但数值不能直接解释为完整历史成本。 +- pending 超时仅告警，未实现按订单号主动远程查询。提交意图落盘后、请求真正发出前崩溃，也需要人工核实；这与 T13 已知写盘失败未回滚不同。 +- 终态识别要求系统订单号、正确的本地标识、可汇总申报量和成交价格。如果拒单没有系统订单号，或 QMT 取消态数量语义不同于模型推定，pending 可能无法自动结束。需要以真实回报核对，不能仅凭 mock 宣布兼容。 +- 仅提供单进程互斥，不支持同账户同状态文件的多进程并发写入。 + +## 仍存在的既有问题 + +### T4【高】开仓和补仓缺少统一预算 + +位置：`strategy/trend/boot.py:191`、`open.py:51`、`positions.py:41`。 + +两条线程各自使用资金，开仓每个信号仍按完整 buy_value 计算。补仓的不确定委托只预留本轮估算金额，下轮又按新的 available 开始，未统一扣除未确认订单的潜在占款。市场价格变化也可能超出快照估算。 + +建议：账户级共享预算，明确券商已冻结资金和本地尚未反映的预留，避免漏计或重复扣减。 + +### T5【高】自动撤单不区分策略归属 + +位置：`strategy/trend/order.py:74`。 + +超时撤单遍历全账户订单，没有本地订单前缀/策略归属限制。R5 同时确认手工来源订单会被传给撤单接口。建议防重参考全账户，自动撤单仅限明确归属本策略的订单。 + +### T10【中，策略取舍】止盈门槛会阻断已激活网格回撤 + +位置：`strategy/trend/positions.py:108`、`libs/grid_take_profit.py`。 + +低于 minimum_profit 即返回，不再观察峰值回撤；进程重启峰值也丢失。若要求激活后持续追踪，应保存激活/峰值；若最低利润是硬性卖出门槛，应明确此行为。 + +### T11【中】信号仅启动加载 + +位置：`strategy/trend/boot.py:69`、`libs/signal.py:17`。 + +首次请求失败会得到空结果，此后不刷新；运行中的新增/撤销不生效。建议按业务时效定期刷新，并区分失败与正常空信号。 + +## IPO 当前问题 + +### I1【高】下单结果未检查就写完成锁 + +位置：`strategy/ipo/boot.py:53`。 + +passorder 返回值被忽略，随后无条件写锁。R6 使用 failed 业务响应，仍生成 A.SH.lock。当前服务端通常将失败表达为 HTTP 异常，该路径会被捕获；但 success 仅说明底层调用未抛异常，服务端未验证字符串化的订单引用，客户端也没有确认受理。 + +建议依据真实受理契约核查结果；不确定时保存待核查记录，不直接标记完成。 + +### I2【高】检查锁—下单—写锁不原子，无券商对账 位置：`strategy/ipo/boot.py:49`、`libs/lockfile.py:9`。 -两个调用可同时看到锁不存在并分别提交；下单已受理但响应超时或写锁失败，也不会留下标记，下轮再次提交。函数没有查询已有券商委托，也没有传入稳定的本地订单标识。重复请求是否由柜台拒绝不在本次静态审计可证明范围内。 +并发调用可同时下单；受理后超时/写锁失败会使后续重新提交。建议账户/发行事件级原子占位与幂等标识，不确定结果先核查。 -建议：按账户与发行事件生成幂等标识，提交前原子占位；不确定结果先查询再决定是否重试。 - -### I3. 锁文件未隔离账户和发行事件（忽略） +### I3【高，共享目录场景】标记不区分账户或发行事件 位置：`strategy/ipo/boot.py:49`。 -路径仅为全局数据目录下的 `{stock}.lock`，既没有账户也没有日期/发行标识。多个账户共享目录时，先运行的账户会阻止另一账户申购；代码复用或残留标记也无法区分新的发行事件。 +只有证券代码作为文件名；多账户共享目录时相互阻止申购，旧标记也无法区分新的发行事件。建议加入账户与发行标识。 -建议：至少加入账户与发行事件标识；不要仅依靠无期限的证券代码标记。 - -## IPO：中风险 - -### I4. 证券代码和数值校验不完整（忽略） +### I4【中】数据校验不完整 位置：`strategy/ipo/boot.py:40`。 -`str(None)` 会变成非空的 `None`；价格 `NaN`/正无穷不会被 `<= 0` 拦截；`int(100.9)` 会截断为 100。证券代码还直接用于文件名，路径字符可导致越界路径或非法文件名。单条异常隔离能保住后续候选，但不能保证本条参数正确。 +str(None) 不是空代码；非有限价格可绕过 <=0；int 浮点数量会截断；证券代码直接进入路径。建议有限正价格、严格整数数量、代码格式与路径字符限制。 -建议：证券代码采用明确格式；价格必须为有限正数，数量必须为正整数且满足实际申购规则；构造路径前排除路径分隔符。 +IPO 已正确逐条处理 list[dict]，使用 with Client，候选异常不终止后续候选；本轮未发现这些路径退化。 -已确认有效：IPO 按 `list[dict]` 逐条读取；Client 使用上下文关闭；候选级异常不会终止后续候选。函数实际返回 `None`，文档中“返回成功数量”和“券商对账”的说明与实现不一致。 +## 已确认的有效修复 -## Trend：严重风险 +- T1 原空订单快照丢 pending：已修复；本地待确认记录独立持久化，缺席不删除。T12 是持仓账本清理的另一条路径。 +- T2 原过滤取消子单导致误判：已修复；启动和每轮对账均使用 portfolio.orders 完整数据。 +- T3 原受理后才写订单关联：已修复；当前提交前保存计划与本地 ID。提交前失败的恢复问题另列 T13。 +- T6 下单网络/解码异常与逐证券异常边界已接入。 +- T7 统一 finally：先关闭线程池，再关闭 Client，初始化异常可释放资源。 +- T8 excluded_codes 在开仓与持仓管理均检查。 +- T9 缺失金额导致均价低估：当前保留未定价子单，延后数量/金额记账，不按零金额算均价。 +- busy_keys 与 SimpleCache 都在 busy/place 验证；补仓首次成交计一次，已完成首次接管规则与 expected_qty 持久化。 -### T1. 快照缺席或撤单过滤会不可逆地丢失未决状态 +## 保留配置与性能观察 -位置：`strategy/trend/order.py:60`、`strategy/trend/state.py:144`、`strategy/trend/boot.py:176`。 +此前要求跳过的行为仍保留：大盘恒允许、不足一手强制 100 股、现金安全线仅限开仓、固定亏损档位与未生效的两个风控配置。时间退出条件仍为 >=15:00，交易时间只判断工作日/时段。 -`refresh()` 覆盖 `data`；短暂空快照没有保留本地 pending。更直接的触发是超时订单：调用撤单后立即 `continue`，无论撤单业务结果如何，都不再进入对账数据。`reconcile()` 找不到订单即把 ING 清为空字符串，无持仓还会删除记录；之后只处理 ING 的逻辑无法回写迟到的成交。 +状态查找使用字典/集合，无变化不写盘；但每个新买单都在 OrderBook.mutex 内序列化整个状态文件并同步写盘，同时持有 State 锁。批量下单的耗时随账本大小和磁盘延迟增长；“最高性能”没有基准测试支持。可先测每轮耗时、文件大小、持锁时间，再考虑批量意图预留与一次落盘；不可为减少写入而取消提交前持久化保障。 -`busy_keys` 已保留这些活动订单，能阻止同方向再次提交，但它没有传给状态对账，不能解决此问题。建议保留待确认状态，并将完整订单快照与展示列表分开；只有明确终态才能结束对账。 +## 本轮验证 -## Trend：高风险 +6 组隔离复现已执行：R1 空持仓重置次数、R2 未发送订单残留、R3 HTTP 400 锁定、R4 终态后金额修正忽略、R5 撤单异常中断对账/误触手工委托、R6 IPO failed 响应写锁。 -### T2. 部分成交与取消拆单被过滤成全部完成 - -位置：`strategy/trend/order.py:69`、`strategy/trend/state.py:153`、`strategy/trend/state.py:213`。 - -同一本地订单包含完成子单和取消子单时，取消子单被过滤；剩余全是 56，状态变为 OK，并增加补仓次数。建议用完整快照判定终态，单独累计实际成交；部分成交的记账与是否完整完成应分别处理。 - -### T3. 下单受理后、状态落盘前崩溃会丢失订单关联 - -位置：`strategy/trend/open.py:89`、`strategy/trend/positions.py:181`。 - -先调用 place 成功，再保存状态。响应丢失、进程退出或写盘失败可能使底仓/补仓缺少对应记录。重启后即使券商活动订单暂时阻止重复提交，补仓次数与成交关联仍可能无法恢复。建议提交前持久化意图，保留可核查订单标识；不要求必须采用 SQLite。 - -### T4. 开仓没有账户余额预算，且与补仓并发 （忽略） - -位置：`strategy/trend/boot.py:185`、`strategy/trend/open.py:49`、`strategy/trend/positions.py:42`。 - -开仓循环每次使用完整 buy_value，没有扣减余额；补仓线程独立使用全部可用资金。多个候选或开仓与补仓同时触发时，计划金额可超出账户资金。建议统一预留预算，受理失败时按确定性结果释放。 - -### T5. Trend 会撤销账户内其他来源的超时订单 （忽略） - -位置：`strategy/trend/boot.py:57`、`strategy/trend/order.py:67`。 - -组合接口返回账户订单，refresh 对全部满足状态和时间条件的订单调用 cancel_by_id，没有检查本地订单前缀或策略归属。若该账户同时有手工、IPO 或其他策略的可撤委托，也会被处理。建议防重可以参考全账户委托，但自动撤单只作用于明确归属本策略的订单。 - -## Trend：中风险及需要确认的行为 - -- **T6 异常隔离不足（已处理）**：下单入口现捕获 APIError、httpx.RequestError 和 SDK 解码抛出的 ValueError，记录异常后返回 False；结果不确定时保留缓存防重，不自动重试。持仓循环增加逐证券 Exception 边界，记录证券代码和堆栈后继续下一只。已通过隔离断言验证连接错误、读取超时、解码错误，以及首只证券异常后第二只仍被处理；未新增 tests 文件。 -- **T7 初始化资源释放（已处理）**：StartTrend 在 Client 创建后使用统一 try/finally，线程池单独持有引用，退出时先等待已创建的线程池结束，再关闭 Client；即使线程池关闭抛异常，内层 finally 仍关闭 Client。已通过 7 个隔离场景验证：组合查询、撤单刷新、状态读取、信号加载、Runtime 构建失败，以及正常退出和线程池关闭异常。未连接柜台或新增测试文件。 -- **T8 排除名单只约束持仓管理（已处理）**：excluded_codes 同时禁止新开仓；open_signal 在每条信号处理前检查并记录跳过原因，与持仓管理一致。已通过隔离验证：高于昨收直接开仓、反弹确认开仓两条路径均跳过排除证券，后续允许证券仍正常处理。未新增测试文件或调用真实交易接口。 -- **T9 成交价缺失时均价可能被低估 （忽略）**：`state.py:225` 对所有成交计入数量，却只对有金额/价格的记录计入金额；例如两笔各 100 股，只有一笔有 1000 元金额，会得到 5 元均价。应标记数据不完整并补查，不把未知金额当作零。当前交易决策主要使用券商 position.open_price，本项直接影响本地记录。 -- **T10 峰值止盈有门槛限制且不持久化 （忽略）**：`positions.py:105` 在低于最低利润时直接返回。已达峰值后跳跌到门槛以下不会触发回撤卖出；进程重启也丢失历史峰值。若目标是“激活后持续追踪”，需保留激活状态和峰值；若门槛是最低可接受卖价，应明确记录这一取舍。 -- **T11 信号仅启动读取 （忽略）**：`boot.py:67` 不在循环内刷新；首次读取失败会使该信号整次运行缺席，后续更新不会生效。建议按业务需要定时刷新。 - -## 当前保留的风控设置 - -以下行为仍存在；按此前明确跳过的决定列为保留风险，不视为本次自动修复授权： - -- `libs/market.py:34` 始终允许开仓/补仓，市场失败或下跌不阻断。 -- `libs/calc.py:10` 不足一手预算仍返回 100 股，可超过单笔额度。 -- `boot.py:147` 现金安全线只限制开仓，未扣除补仓安全储备。 -- `positions.py:16` 固定亏损档位，最低盈利按股价计算，配置的 loss_trigger_pct/min_profit_pct 不参与 Trend。 - -另：`boot.py:90` 当前仍为 `>= 15:00:00`，与此前“大于 15:00”约定不一致。IPO 与 Trend 使用的 trading_time 仅校验工作日和时段，不包含节假日日历；本次未扩展该功能。 - -## 已确认修复与验证边界 - -- Trend 的 busy() 和 place() 均检查 busy_keys + SimpleCache。集合在撤单前收集，故请求撤单仍保留防重；refresh 不会续期 SimpleCache。 -- Trend 调用现有 passorder，SDK 参数对应一致。 -- 未知信号配置通过 continue 跳过，不再解引用 None。 -- 旧测试共 6 个 `.py` 文件按要求删除，可从 Git 恢复；不将旧测试失败统计当成本轮结果，也不重新建立 tests 文件。 -- 本次只进行语法与隔离行为检查，未验证真实柜台受理、撤单终态和回报字段。服务端直接字符串化底层 passorder 返回值，Trend 以有效 order_ref 判断成功；二者能否匹配真实 QMT 必须以实际回报核验，不能仅凭静态代码断言。 - -优先处理 T1/T2 对账数据问题、I1/I2 申购确认与幂等、T5 撤单归属，再评估资金预算与其他保留风险。 +R4 属于有条件的上游契约风险；其余复现验证代码在所述输入/故障下的行为，不等价于断言柜台必然出现该故障。没有新增 tests 文件或真实网络交易请求。 diff --git a/docs/trend-state.md b/docs/trend-state.md new file mode 100644 index 0000000..2244dce --- /dev/null +++ b/docs/trend-state.md @@ -0,0 +1,37 @@ +# Trend 精简状态记录（JSON v3） + +## 记录内容 + +- `items[code]`：底仓数量、底仓均价、补仓次数、累计补仓数量和金额。补仓均价由金额除以数量得到。 +- `pending[code]`：每只证券最多一个待确认买单，仅存订单号、类型（base/add）、预开仓数量 expected_qty 和提交时间。 +- 文件仍为 `{strategy}_{account_id}_state.json`，内存字典访问，数据变化时原子写 JSON。 + +没有子单历史缓存、增量游标、counted 标记、待确认计数器或 seen_positions。 + +## 处理规则 + +1. 首次接管的持仓全部作为底仓；存在 pending 的证券等待自身订单对账，不从部分持仓重复导入。 +2. 提交前保存 pending 和预开仓数量。保存失败回滚占位，阻止发送；异常响应保留待确认状态。 +3. 订单处理中不更新账本。空快照、部分快照和缺少成交价格时继续等待；超过 180 秒定期提示核查，不自动释放。 +4. 当前完整快照必须覆盖预开仓数量，所有子单明确结束，且成交金额完整，才一次记账。系统订单号用于本轮去重，不跨轮缓存子单。 +5. 有成交的补仓计一次，累加实际数量和金额；部分成交后撤单也如此。完全未成交不增加补仓次数。 +6. 更新账本和删除 pending 在同一次文件替换中保存，重复快照/重启不会重复记账。 +7. 账本不因持仓缺席而删除。清仓后若需要同证券重新开底仓，必须先核实清仓、停机备份并清理对应 items 记录；当前未实现自动清仓判定。 + +SimpleCache 与 busy_keys 继续用于快速防重，本地 pending 同样阻止买入。底仓数量/补仓数量是买入成交记录，不自动分摊卖出；实际可用持仓仍以券商为准。 + +## 迁移 + +自动支持 v2 → v3，保存前备份为 `.json.v2.bak`。 + +v2 尚未结束的订单可能已经记入部分成交和一次补仓计数。迁移先撤回该 pending 的 applied_qty、applied_amount、counted 对账本的贡献；最终完整快照到达后再一次计入，避免重复。已完成订单的账本保持原值。原子保存失败时原 v2 文件仍可重试。 + +v2 子单缓存保存在备份中，v3 不继续使用；需要之后提供完整订单快照才能结束对账。expected_qty=0 的旧迁移记录仍需要核实计划量。若同一证券存在多个旧 pending，停止迁移并保留原文件，避免静默丢掉订单。 + +更早的无版本 JSON 不再由精简状态机自动猜测转换，会报错且不修改原文件，需先核实转换。旧文件未保存的历史累计补仓成本无法恢复。 + +## 验证与限制 + +已使用临时目录和替身验证：首次接管、处理中不记账、部分成交撤单、完整拆单、重复记录/重启、空持仓保留、累计成本、无变化不写盘、写盘失败回滚、拒单/零成交清理，以及 v2 增量撤回后只记一次终态成交。没有新增 tests 文件，也没有真实交易调用。 + +依赖柜台提供完整终态快照、正确的本地订单标识及申报量/成交字段。超时只提示人工核查，不自动查询或重发。单进程使用，未提供多进程文件互斥。 diff --git a/py-client/strategy/trend/boot.py b/py-client/strategy/trend/boot.py index d60c98d..f84ee0b 100644 --- a/py-client/strategy/trend/boot.py +++ b/py-client/strategy/trend/boot.py @@ -56,15 +56,14 @@ def StartTrend() -> None: portfolio = client.portfolio() assets = portfolio.assets positions = list(portfolio.positions.values()) - order_book = OrderBook() - order_book.refresh(client, portfolio.orders) - storeState = State.for_strategy( config.global_config.qmt_data_dir, config.account_config.strategy, config.account_config.account_id, ) - storeState.reconcile(positions, order_book.data) + order_book = OrderBook(state=storeState) + order_book.refresh(client, portfolio.orders) + storeState.reconcile(positions, portfolio.orders) # 获取本策略的信号开仓数据 signals = init_signals( @@ -180,7 +179,7 @@ def RunOnce(run: Runtime, signals:list[SignalItem]) -> None: # 6. 更新状态机 try: - run.state.reconcile(positions, run.orders.data) + run.state.reconcile(positions, portfolio.orders) except Exception: log.exception("[状态] 订单状态对账失败") return diff --git a/py-client/strategy/trend/open.py b/py-client/strategy/trend/open.py index 9c3966f..255c76b 100644 --- a/py-client/strategy/trend/open.py +++ b/py-client/strategy/trend/open.py @@ -8,7 +8,6 @@ from libs import calc_buy_volume from sdk import OP_BUY from .runtime import Runtime from .order import PlaceOrderRequest -from .state import STATUS_ING, StateItem import logging as log @@ -88,16 +87,11 @@ def do_open(run: Runtime, code: str, volume: int, signal_key: str, price: float) volume, order_id, signal_key, + kind="base", ) if not run.orders.place(request): raise RuntimeError("订单提交失败") - run.state.set(StateItem( - code=code, - base_order_id=order_id, - base_status=STATUS_ING, - )) - run.state.save() run.open_watch.forget(code) diff --git a/py-client/strategy/trend/order.py b/py-client/strategy/trend/order.py index 5b49d39..b69bb5a 100644 --- a/py-client/strategy/trend/order.py +++ b/py-client/strategy/trend/order.py @@ -3,15 +3,14 @@ from __future__ import annotations import secrets -import logging as log from dataclasses import dataclass from datetime import datetime, timedelta from threading import Lock -from typing import Any from cachelib import SimpleCache -import httpx +import logging -from sdk import APIError, ORDER_SIDE_BY_OFFSET, Client, OrderItem +import httpx +from sdk import Client,ORDER_SIDE_BY_OFFSET,APIError,OrderItem # 表示委托仍在处理、可能继续成交的 QMT 状态。 BUSY_STATUSES = {"48", "49", "50", "51", "52", "55"} @@ -23,13 +22,12 @@ CANCELABLE_STATUSES = {"49", "50", "51", "52"} @dataclass(frozen=True, slots=True) class PlaceOrderRequest: """``OrderBook.place`` 提交委托所需的全部参数。""" - - client: Any op: int code: str volume: int order_id: str strategy_name: str + kind: str = "" class OrderBook: @@ -79,30 +77,28 @@ class OrderBook: ): client.cancel_by_id(item.id) canceled += 1 - log.info("[Order] 超时撤单，代码=%s，方向=%s，柜台订单=%s", item.code, item.side, item.id) + logging.info("[Order] 超时撤单，代码=%s，方向=%s，柜台订单=%s", item.code, item.side, item.id) continue # 缓存本次有效订单 data.append(item) - - with self.mutex: self.data = data self.busy_keys = busy_keys - log.info("[Order] 刷新完成，跟踪=%d，处理中=%d，撤销=%d", len(data), len(busy_keys), canceled) + logging.info("[Order] 刷新完成，跟踪=%d，处理中=%d，撤销=%d", len(data), len(busy_keys), canceled) def place(self, request: PlaceOrderRequest) -> bool: """按最新价提交委托，并立即写入本地方向锁。""" side = ORDER_SIDE_BY_OFFSET.get(str(request.op), "") if not side: - log.warning("[Order] 下单失败，代码=%s，原因=未知买卖方向(%s)", request.code, request.op) + logging.warning("[Order] 下单失败，代码=%s，原因=未知买卖方向(%s)", request.code, request.op) return False key = self._busy_key(side, request.code) with self.mutex: if key in self.busy_keys or self.busy_cache.has(key): - log.info("[Order] 跳过重复下单，代码=%s，方向=%s", request.code, side) + logging.info("[Order] 跳过重复下单，代码=%s，方向=%s", request.code, side) return False self.busy_cache.set(key, True, timeout=self.lock_timeout_sec) @@ -115,31 +111,12 @@ class OrderBook: order_id=request.order_id, ) except APIError as exc: - log.exception("[Order] 下单失败，代码=%s，本地订单=%s，HTTP状态=%d，错误=%s", request.code, request.order_id, exc.status_code, exc.message or str(exc)) + logging.exception("[Order] 下单失败，代码=%s，本地订单=%s，HTTP状态=%d，错误=%s", request.code, request.order_id, exc.status_code, exc.message or str(exc)) return False except (httpx.RequestError, ValueError): # 响应异常不能证明柜台未受理，保留缓存防重，不自动重试。 - log.exception("[Order] 下单请求或响应异常，代码=%s，本地订单=%s", request.code, request.order_id) + logging.exception("[Order] 下单请求或响应异常，代码=%s，本地订单=%s", request.code, request.order_id) return False - if not isinstance(result, dict): - log.warning("[Order] 下单失败，代码=%s，本地订单=%s，原因=响应格式无效", request.code, request.order_id) - return False - order_ref = str(result.get("order_ref") or "").strip().lower() - if result.get("status") != "success" or order_ref in {"", "unknown", "none"}: - log.warning("[Order] 下单被拒绝，代码=%s，本地订单=%s，状态=%s，柜台订单=%s", request.code, request.order_id, result.get("status"), order_ref) - return False - - pending = OrderItem( - id=order_ref, - code=request.code, - side=side, - remark=request.order_id, - status="48", - created_at=datetime.now(), - volume=request.volume, - local_order_id=request.order_id, - ) - with self.mutex: - self.data.append(pending) - log.info("[Order] 下单已受理，代码=%s，方向=%s，数量=%d，本地订单=%s，柜台订单=%s", request.code, side, request.volume, request.order_id, order_ref) + + logging.info("[Order] 下单已受理，代码=%s，方向=%s，数量=%d，本地订单=%s，返回=%s", request.code, side, request.volume, request.order_id, result) return True diff --git a/py-client/strategy/trend/positions.py b/py-client/strategy/trend/positions.py index 237c77b..bc6205e 100644 --- a/py-client/strategy/trend/positions.py +++ b/py-client/strategy/trend/positions.py @@ -10,7 +10,6 @@ from sdk import OP_BUY, OP_SELL, PositionItem, Tick from .order import PlaceOrderRequest from .runtime import Runtime -from .state import STATUS_ING import logging as log LOSS_TIERS = (-30.0, -50.0) @@ -181,14 +180,11 @@ def handle_loss( volume=volume, order_id=order_id, strategy_name=runtime.account_cfg.strategy, + kind="add", ) if not runtime.orders.place(request): - return TradeDecision(False, "补仓委托失败") - - state.added_status = STATUS_ING - state.added_order_id = order_id - runtime.state.set(state) - runtime.state.save() + reserved = amount if runtime.state.busy(position.stock_code) else 0.0 + return TradeDecision(False, "补仓委托失败或待确认", reserved) runtime.add_watch.forget(position.stock_code) return TradeDecision(True, f"买入 {volume} 股，订单={order_id}", amount) diff --git a/py-client/strategy/trend/state.py b/py-client/strategy/trend/state.py index 81740af..7351e3a 100644 --- a/py-client/strategy/trend/state.py +++ b/py-client/strategy/trend/state.py @@ -1,239 +1,207 @@ -"""趋势策略持仓状态的内存管理与 JSON 持久化。""" +"""简单成交账本：订单明确结束后一次记账，JSON 原子保存。""" from __future__ import annotations import json import logging as log -from dataclasses import asdict, dataclass +import math +import shutil +import time +from dataclasses import asdict, dataclass, field, replace from pathlib import Path -from threading import Lock +from threading import RLock from typing import Iterable from sdk import OrderItem, PositionItem -from .order import BUSY_STATUSES, COMPLETED_STATUSES - -# 委托状态：无操作、处理中、已完成。 -STATUS_NONE = "" -STATUS_ING = "ING" -STATUS_OK = "OK" +TERMINAL_STATUSES = {"53", "54", "56", "57"} @dataclass(slots=True) class StateItem: - """单只证券的底仓和补仓状态。""" - - # 证券代码。 code: str - - # 底仓订单、数量、成本和处理状态。 base_order_id: str = "" + base_pre_qty: int =0 base_qty: int = 0 base_cost: float = 0.0 - base_status: str = STATUS_NONE - - # 补仓订单、补仓次数、数量、成本和处理状态。 added_order_id: str = "" + added_pre_qty: int =0 added_num: int = 0 added_qty: int = 0 - added_cost: float = 0.0 - added_status: str = STATUS_NONE + added_amount: float = 0.0 + + @property + def added_cost(self) -> float: + return self.added_amount / self.added_qty if self.added_qty else 0.0 + + +@dataclass(slots=True) +class PendingOrder: + order_id: str + kind: str + expected_qty: int + submitted_at: float = field(default_factory=time.time) class State: - """线程安全的策略状态存储。 - - 状态以内存字典提供快速访问，并通过临时文件替换的方式写入 JSON， - 防止程序在写入过程中退出而破坏原状态文件。 - """ - def __init__(self, path: str | Path) -> None: self.path = Path(path) - self.lock = Lock() - self.items = self._load() + self.lock = RLock() + self.items: dict[str, StateItem] = {} + self.pending: dict[str, PendingOrder] = {} # 每只证券最多一个待确认买单 + self._dirty = False + self._next_warning = 0.0 + self._load() @classmethod - def for_strategy( - cls, - data_dir: str | Path, - strategy: str, - account_id: str, - ) -> "State": - """根据数据目录、策略名称和账户生成独立状态文件。""" - state_path = Path(data_dir) / f"{strategy}_{account_id}_state.json" - return cls(state_path) + def for_strategy(cls, data_dir: str | Path, strategy: str, account_id: str) -> State: + return cls(Path(data_dir) / f"{strategy}_{account_id}_state.json") @property def codes(self) -> list[str]: - """返回当前已经接管的全部证券代码快照。""" with self.lock: return list(self.items) def get(self, code: str) -> StateItem: - """获取指定证券的状态；不存在时抛出 KeyError。""" with self.lock: - return self.items[code] + return replace(self.items[code]) - def set(self, item: StateItem) -> None: - """新增或覆盖一只证券的状态。""" + def busy(self, code: str) -> bool: with self.lock: - self.items[item.code] = item + return code in self.pending - def delete(self, code: str) -> bool: - """删除已终结的证券状态，并返回是否实际删除。""" + def begin(self, order_id: str, code: str, kind: str, expected_qty: int) -> bool: + if kind not in {"base", "add"} or type(expected_qty) is not int or expected_qty <= 0: + raise ValueError("订单类型或预开仓数量无效") with self.lock: - return self.items.pop(code, None) is not None + if code in self.pending or (kind == "base" and code in self.items): + return False + if kind == "add" and code not in self.items: + raise ValueError("缺少底仓记录") + dirty = self._dirty + self.pending[code] = PendingOrder(order_id, kind, expected_qty) + self._dirty = True + try: + self.save() # 必须成功落盘后才允许发送请求 + except Exception: + self.pending.pop(code) + self._dirty = dirty + raise + return True + def reject(self, order_id: str) -> None: + """仅用于已明确未受理的请求。""" + with self.lock: + for code, pending in self.pending.items(): + if pending.order_id == order_id: + del self.pending[code] + self._dirty = True + self.save() + return - def sync_positions(self, positions: Iterable[PositionItem]) -> None: - """把尚未接管的真实持仓初始化为已完成底仓。 - - 无证券代码、无持仓数量或成本无效的记录会被忽略。同步结束后 - 立即保存，确保首次接管的持仓在程序重启后仍可恢复。 - """ - known_codes = set(self.codes) - imported = 0 - for position in positions: - if ( - not position.stock_code - or position.volume <= 0 - or position.open_price <= 0 - or position.stock_code in known_codes - ): - continue - - self.set( - StateItem( - base_order_id=position.trade_id, - code=position.stock_code, - base_qty=position.volume, - base_cost=round(position.open_price, 2), - base_status=STATUS_OK, - ) - ) - known_codes.add(position.stock_code) - imported += 1 - - self.save() - if imported: - log.info("[状态] 导入持仓=%d，状态总数=%d", imported, len(known_codes)) - - def reconcile( - self, - positions: Iterable[PositionItem], - orders: list[OrderItem], - ) -> None: - """用真实持仓和委托恢复本地状态；拆分订单全部完成才算完成。""" - position_list = list(positions) - self.sync_positions(position_list) - position_codes = { - item.stock_code for item in position_list if item.volume > 0 - } - orders_by_local_id: dict[str, list[OrderItem]] = {} + def reconcile(self, positions: Iterable[PositionItem], orders: list[OrderItem]) -> None: + by_id: dict[str, dict[str, OrderItem]] = {} for order in orders: - if order.local_order_id: - orders_by_local_id.setdefault(order.local_order_id, []).append(order) + if order.local_order_id and order.id: + by_id.setdefault(order.local_order_id, {})[order.id] = order + with self.lock: + for position in positions: + code = position.stock_code + if (code and code not in self.items and code not in self.pending + and position.volume > 0 and math.isfinite(position.open_price) + and position.open_price > 0): + self.items[code] = StateItem(code, position.volume, position.open_price) + self._dirty = True - for code in list(self.codes): - item = self.get(code) - for order_id_attr, status_attr, qty_attr, cost_attr in ( - ("base_order_id", "base_status", "base_qty", "base_cost"), - ("added_order_id", "added_status", "added_qty", "added_cost"), - ): - local_order_id = getattr(item, order_id_attr) - current_status = getattr(item, status_attr) - if current_status != STATUS_ING or not local_order_id: + now = time.time() + for code, pending in list(self.pending.items()): + rows = [o for o in by_id.get(pending.order_id, {}).values() + if o.code == code and o.side == "BUY"] + filled = _finished_fill(rows, pending.expected_qty) + if filled is None: + if now >= self._next_warning and now - pending.submitted_at >= 180: + log.warning("[状态] 订单待核查，代码=%s，订单=%s，计划=%d；保留防重", + code, pending.order_id, pending.expected_qty) continue - - matching_orders = orders_by_local_id.get(local_order_id) - status = _order_status(matching_orders) - if status != current_status: - log.info("[状态] %s 订单=%s，状态=%s->%s", code, local_order_id, current_status, status) - setattr(item, status_attr, status) - if status == STATUS_OK: - quantity, cost = _filled_order(matching_orders) - if quantity > 0: - setattr(item, qty_attr, quantity) - if cost > 0: - setattr(item, cost_attr, cost) - if status_attr == "added_status": + qty, amount = filled + if qty: + item = self.items.setdefault(code, StateItem(code)) + if pending.kind == "base": + item.base_qty, item.base_cost = qty, amount / qty + else: item.added_num += 1 - self.set(item) - - # Opening orders normally have no position until their first fill. Order - # reconciliation must therefore happen before stale state is removed. - for code in list(self.codes): - if code not in position_codes: - item = self.get(code) - if any( - order_id and order_id in orders_by_local_id - for order_id in (item.base_order_id, item.added_order_id) - ): - continue - if self.delete(code): - log.info("[状态] 已移除持仓状态，代码=%s", code) - self.save() + item.added_qty += qty + item.added_amount += amount + del self.pending[code] + self._dirty = True + log.info("[状态] 对账结束，代码=%s，订单=%s，计划=%d，成交=%d", + code, pending.order_id, pending.expected_qty, qty) + if now >= self._next_warning: + self._next_warning = now + 180 + self.save() # 成交记账与移除 pending 在同一文件中原子提交 def save(self) -> None: - """将内存状态格式化写入 JSON，并原子替换正式文件。""" with self.lock: + if not self._dirty: + return self.path.parent.mkdir(parents=True, exist_ok=True) - temporary_path = self.path.with_suffix(self.path.suffix + ".tmp") - payload = { - code: asdict(item) - for code, item in self.items.items() - } - temporary_path.write_text( - json.dumps(payload, ensure_ascii=False, indent=2) + "\n", - encoding="utf-8", - ) - temporary_path.replace(self.path) + temporary = self.path.with_suffix(self.path.suffix + ".tmp") + payload = dict(version=3, items={k: asdict(v) for k, v in self.items.items()}, + pending={k: asdict(v) for k, v in self.pending.items()}) + temporary.write_text(json.dumps(payload, ensure_ascii=False, indent=2, allow_nan=False) + "\n", encoding="utf-8") + temporary.replace(self.path) + self._dirty = False - def _load(self) -> dict[str, StateItem]: - """读取已有状态文件；文件不存在时从空状态开始。""" + def _load(self) -> None: try: raw = json.loads(self.path.read_text(encoding="utf-8")) except FileNotFoundError: - return {} - except (OSError, json.JSONDecodeError) as exc: - raise ValueError(f"[状态] 读取或解析失败: {exc}") from exc - - if not isinstance(raw, dict): - raise ValueError("[状态] 状态文件根节点必须是 JSON 对象") - - try: - return { - code: StateItem(**item) - for code, item in raw.items() - } - except (TypeError, ValueError) as exc: - raise ValueError(f"[状态] 状态字段无效: {exc}") from exc + return + if raw.get("version") not in {2, 3}: + raise ValueError("仅支持 v2/v3 状态文件；旧格式请先核实转换，原文件未修改") + self.items = {k: StateItem(**v) for k, v in raw["items"].items()} + if raw["version"] == 3: + self.pending = {k: PendingOrder(**v) for k, v in raw["pending"].items()} + return + # v2 的未决订单可能已经增量记账，先撤回这部分，结束时再完整记一次。 + for old in raw["pending"].values(): + code = old["code"] + if code in self.pending: + raise ValueError(f"{code} 存在多个旧未决订单，请核查；原文件未修改") + item = self.items[code] + qty, amount = old.get("applied_qty", 0), old.get("applied_amount", 0.0) + if old["kind"] == "add": + item.added_qty -= qty + item.added_amount -= amount + item.added_num -= int(old.get("counted", False)) + else: + base_amount = item.base_qty * item.base_cost - amount + item.base_qty -= qty + item.base_cost = base_amount / item.base_qty if item.base_qty else 0.0 + if not item.base_qty and not item.added_qty: + del self.items[code] + self.pending[code] = PendingOrder(old["order_id"], old["kind"], old["expected_qty"], + old.get("submitted_at", time.time())) + backup = self.path.with_suffix(self.path.suffix + ".v2.bak") + if not backup.exists(): + shutil.copy2(self.path, backup) + self._dirty = True + self.save() -def _order_status(orders: list[OrderItem] | None) -> str: - """将柜台订单简化为无状态、处理中或已成交。""" - if not orders: - return STATUS_NONE - statuses = {order.status for order in orders} - if statuses <= COMPLETED_STATUSES: - return STATUS_OK - if statuses <= BUSY_STATUSES | COMPLETED_STATUSES: - return STATUS_ING - return STATUS_NONE - - -def _filled_order(orders: list[OrderItem] | None) -> tuple[int, float]: - """汇总已成交订单的实际数量和加权成交价。""" - quantity = 0 - amount = 0.0 - for order in orders or []: - filled = order.traded_volume if order.traded_volume > 0 else order.volume - if filled <= 0: - continue - quantity += filled - if order.trade_amount > 0: - amount += order.trade_amount - elif order.trade_price > 0: - amount += order.trade_price * filled - cost = round(amount / quantity, 4) if quantity > 0 and amount > 0 else 0.0 - return quantity, cost +def _finished_fill(orders: list[OrderItem], expected_qty: int) -> tuple[int, float] | None: + """仅完整终态快照可记账；缺项、未成交完或金额未知均继续等待。""" + if not orders or expected_qty <= 0 or sum(o.volume for o in orders) != expected_qty: + return None + qty, amount = 0, 0.0 + for order in orders: + if (order.status not in TERMINAL_STATUSES or not 0 <= order.traded_volume <= order.volume + or (order.status == "56" and order.traded_volume != order.volume)): + return None + if order.traded_volume: + value = order.trade_amount if order.trade_amount > 0 else order.trade_price * order.traded_volume + if not math.isfinite(value) or value <= 0: + return None + qty += order.traded_volume + amount += value + return qty, amount