From 63ffc1329bc747e50b46b3142ec80cd7b6880014 Mon Sep 17 00:00:00 2001 From: yanweidong Date: Sat, 5 Sep 2026 11:46:31 +0800 Subject: [PATCH] fix bug --- docs/ipo-trend-audit.md | 109 ++++++++ docs/trend-audit.md | 2 + py-client/strategy/ipo/boot.py | 2 +- py-client/strategy/trend/boot.py | 115 +++++---- py-client/strategy/trend/open.py | 3 + py-client/strategy/trend/order.py | 5 + py-client/strategy/trend/positions.py | 93 +++---- py-client/tests/test_ipo.py | 140 ---------- py-client/tests/test_market.py | 22 -- py-client/tests/test_signal.py | 33 --- py-client/tests/test_trend.py | 359 -------------------------- py-client/tests/test_trend_busy.py | 42 --- py-client/tests/test_zt.py | 35 --- 13 files changed, 229 insertions(+), 731 deletions(-) create mode 100644 docs/ipo-trend-audit.md delete mode 100644 py-client/tests/test_ipo.py delete mode 100644 py-client/tests/test_market.py delete mode 100644 py-client/tests/test_signal.py delete mode 100644 py-client/tests/test_trend.py delete mode 100644 py-client/tests/test_trend_busy.py delete mode 100644 py-client/tests/test_zt.py diff --git a/docs/ipo-trend-audit.md b/docs/ipo-trend-audit.md new file mode 100644 index 0000000..5817c68 --- /dev/null +++ b/docs/ipo-trend-audit.md @@ -0,0 +1,109 @@ +# IPO / Trend 当前代码审计 + +日期：2026-09-05。范围：`py-client/strategy/{ipo,trend}`，以及直接依赖的 SDK、公共库、配置与 `api/qmt_rest_new.py` 下单/撤单处理器。以下路径除特别注明外均相对于 `py-client/`。 + +本报告取代上一轮 Trend 审计的当前结论。仅清理测试文件、更新文档；未实施策略修复。采用静态阅读与隔离的内存断言，不连接柜台、不提交真实订单。风险等级按状态丢失、重复委托、漏单等后果划分；未验证真实 QMT 的返回值与回报时序。 + +## IPO：高风险 + +### I1. 未确认受理就写入永久申购标记 （忽略） + +位置：`strategy/ipo/boot.py:53`、`libs/lockfile.py:14`；服务端 `api/qmt_rest_new.py:246`。 + +`client.passorder()` 返回值被忽略，随后无条件写锁、记录申购日志。服务端只要底层调用未抛异常就返回 success，并直接把返回值字符串化，未验证引用是否有效。因此本地标记只能证明调用返回，不能证明委托有效。遇到失败业务响应或无效引用，后续轮次仍因锁文件存在而跳过，导致漏申购。 + +建议：依据明确的柜台受理契约验证结果；无法确认的结果保留待核查记录，按委托/成交查询确认后再标记完成。 + +### I2. 检查锁、下单、写锁不是原子操作，且没有券商对账（忽略） + +位置：`strategy/ipo/boot.py:49`、`libs/lockfile.py:9`。 + +两个调用可同时看到锁不存在并分别提交；下单已受理但响应超时或写锁失败，也不会留下标记，下轮再次提交。函数没有查询已有券商委托，也没有传入稳定的本地订单标识。重复请求是否由柜台拒绝不在本次静态审计可证明范围内。 + +建议：按账户与发行事件生成幂等标识，提交前原子占位；不确定结果先查询再决定是否重试。 + +### I3. 锁文件未隔离账户和发行事件（忽略） + +位置：`strategy/ipo/boot.py:49`。 + +路径仅为全局数据目录下的 `{stock}.lock`，既没有账户也没有日期/发行标识。多个账户共享目录时，先运行的账户会阻止另一账户申购；代码复用或残留标记也无法区分新的发行事件。 + +建议：至少加入账户与发行事件标识；不要仅依靠无期限的证券代码标记。 + +## IPO：中风险 + +### I4. 证券代码和数值校验不完整（忽略） + +位置：`strategy/ipo/boot.py:40`。 + +`str(None)` 会变成非空的 `None`；价格 `NaN`/正无穷不会被 `<= 0` 拦截；`int(100.9)` 会截断为 100。证券代码还直接用于文件名，路径字符可导致越界路径或非法文件名。单条异常隔离能保住后续候选，但不能保证本条参数正确。 + +建议：证券代码采用明确格式；价格必须为有限正数，数量必须为正整数且满足实际申购规则；构造路径前排除路径分隔符。 + +已确认有效：IPO 按 `list[dict]` 逐条读取；Client 使用上下文关闭；候选级异常不会终止后续候选。函数实际返回 `None`，文档中“返回成功数量”和“券商对账”的说明与实现不一致。 + +## Trend：严重风险 + +### T1. 快照缺席或撤单过滤会不可逆地丢失未决状态 + +位置：`strategy/trend/order.py:60`、`strategy/trend/state.py:144`、`strategy/trend/boot.py:176`。 + +`refresh()` 覆盖 `data`；短暂空快照没有保留本地 pending。更直接的触发是超时订单：调用撤单后立即 `continue`，无论撤单业务结果如何，都不再进入对账数据。`reconcile()` 找不到订单即把 ING 清为空字符串，无持仓还会删除记录；之后只处理 ING 的逻辑无法回写迟到的成交。 + +`busy_keys` 已保留这些活动订单，能阻止同方向再次提交，但它没有传给状态对账，不能解决此问题。建议保留待确认状态，并将完整订单快照与展示列表分开；只有明确终态才能结束对账。 + +## Trend：高风险 + +### 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 撤单归属，再评估资金预算与其他保留风险。 diff --git a/docs/trend-audit.md b/docs/trend-audit.md index e30ab94..25eae00 100644 --- a/docs/trend-audit.md +++ b/docs/trend-audit.md @@ -1,5 +1,7 @@ # Trend 策略代码审计报告 +> 历史报告：当前 IPO / Trend 复核结果见 [ipo-trend-audit.md](ipo-trend-audit.md)。下述问题状态与测试统计不代表当前版本；尤其 busy_keys 防重已实现，旧测试文件已按要求清理。 + 审计日期：2026-09-05 审计对象：`py-client/strategy/trend` 当前工作树版本。 关联范围：仅核对直接影响 Trend 行为的 `sdk`、`libs`、`config` 和 Trend 测试。 diff --git a/py-client/strategy/ipo/boot.py b/py-client/strategy/ipo/boot.py index 7de8033..9e98c2e 100644 --- a/py-client/strategy/ipo/boot.py +++ b/py-client/strategy/ipo/boot.py @@ -15,7 +15,7 @@ IPO_SESSIONS = ((time(9, 30), time(11, 30)), (time(13, 0), time(15, 0))) def AutoBuyIpo() -> None: - """安全执行一次新股申购，返回成功提交的证券数量。""" + """安全执行一次新股申购。""" if not config.account_config.enable_auto_ipo: logging.info("[IPO] 自动申购未启用") return diff --git a/py-client/strategy/trend/boot.py b/py-client/strategy/trend/boot.py index 9afab61..d60c98d 100644 --- a/py-client/strategy/trend/boot.py +++ b/py-client/strategy/trend/boot.py @@ -51,65 +51,72 @@ def StartTrend() -> None: config.global_config.qmt_token, config.HTTP_TIMEOUT, ) - portfolio = client.portfolio() - assets = portfolio.assets - positions = list(portfolio.positions.values()) - order_book = OrderBook() - order_book.refresh(client, portfolio.orders) + executor = None + try: + 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) + 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) - # 获取本策略的信号开仓数据 - signals = init_signals( - config.global_config, - config.account_config.signal_allow, - ) - log.info("[启动] 趋势策略已启动，账户=%s，信号=%d，持仓=%d", config.account_config.account_id, len(signals), len(positions)) - run = Runtime( - client=client, - global_cfg=config.global_config, - account_cfg=config.account_config, - state=storeState, - orders=order_book, - open_watch=DipWatch(), - add_watch=DipWatch(), - profit_tracker=GridTrailingTracker(config.account_config.grid_step_pct), - executor=ThreadPoolExecutor(max_workers=3, thread_name_prefix="trend"), - ) + # 获取本策略的信号开仓数据 + signals = init_signals( + config.global_config, + config.account_config.signal_allow, + ) + log.info("[启动] 趋势策略已启动，账户=%s，信号=%d，持仓=%d", config.account_config.account_id, len(signals), len(positions)) + executor = ThreadPoolExecutor(max_workers=3, thread_name_prefix="trend") + run = Runtime( + client=client, + global_cfg=config.global_config, + account_cfg=config.account_config, + state=storeState, + orders=order_book, + open_watch=DipWatch(), + add_watch=DipWatch(), + profit_tracker=GridTrailingTracker(config.account_config.grid_step_pct), + executor=executor, + ) - Overview(assets, positions, config.account_config) + Overview(assets, positions, config.account_config) - DEFAULT_TICK_INTERVAL = 30 - while True: - lt = time.localtime() - if (lt.tm_hour, lt.tm_min, lt.tm_sec) >= (15, 0, 0): - log.info("[Trend] 已到 15:00，结束趋势策略") - run.client.close() - run.executor.shutdown() - return - current_sec = lt.tm_sec - - # 计算距离下一个目标时间点（0秒或30秒）的等待时间 - if current_sec < DEFAULT_TICK_INTERVAL: - wait_seconds = DEFAULT_TICK_INTERVAL - current_sec - elif current_sec < 60: - wait_seconds = 60 - current_sec - else: - wait_seconds = DEFAULT_TICK_INTERVAL - - # 等待到目标时间点 - time.sleep(wait_seconds) - - # 单轮失败不能杀死唯一的交易定时线程。 + DEFAULT_TICK_INTERVAL = 30 + while True: + lt = time.localtime() + if (lt.tm_hour, lt.tm_min, lt.tm_sec) >= (15, 0, 0): + log.info("[Trend] 已到 15:00，结束趋势策略") + return + current_sec = lt.tm_sec + + # 计算距离下一个目标时间点（0秒或30秒）的等待时间 + if current_sec < DEFAULT_TICK_INTERVAL: + wait_seconds = DEFAULT_TICK_INTERVAL - current_sec + elif current_sec < 60: + wait_seconds = 60 - current_sec + else: + wait_seconds = DEFAULT_TICK_INTERVAL + + # 等待到目标时间点 + time.sleep(wait_seconds) + + # 单轮失败不能杀死唯一的交易定时线程。 + try: + RunOnce(run, signals) + except Exception as e: + log.error(f"[Trend] 本 tick 执行失败，下一 tick 继续: {e}", exc_info=True) + finally: try: - RunOnce(run, signals) - except Exception as e: - log.error(f"[Trend] 本 tick 执行失败，下一 tick 继续: {e}", exc_info=True) + if executor is not None: + executor.shutdown(wait=True) + finally: + client.close() def RunOnce(run: Runtime, signals:list[SignalItem]) -> None: diff --git a/py-client/strategy/trend/open.py b/py-client/strategy/trend/open.py index 8b94990..9c3966f 100644 --- a/py-client/strategy/trend/open.py +++ b/py-client/strategy/trend/open.py @@ -15,6 +15,9 @@ import logging as log def open_signal(run:Runtime, ticks, open_signals) -> None: """逐个验证开仓信号并提交买入委托。""" for item in open_signals: + if item.code in run.account_cfg.excluded_codes: + log.info("[Open] %s 信号=%s，跳过：已配置为排除股票", item.code, item.signal_key) + continue # 1. 验证信号配置允许开仓的时间区间。 signal_config = run.global_cfg.signals.get(item.signal_key) if signal_config is None: diff --git a/py-client/strategy/trend/order.py b/py-client/strategy/trend/order.py index 8113596..5b49d39 100644 --- a/py-client/strategy/trend/order.py +++ b/py-client/strategy/trend/order.py @@ -9,6 +9,7 @@ from datetime import datetime, timedelta from threading import Lock from typing import Any from cachelib import SimpleCache +import httpx from sdk import APIError, ORDER_SIDE_BY_OFFSET, Client, OrderItem @@ -116,6 +117,10 @@ class OrderBook: 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)) return False + except (httpx.RequestError, ValueError): + # 响应异常不能证明柜台未受理，保留缓存防重，不自动重试。 + log.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 diff --git a/py-client/strategy/trend/positions.py b/py-client/strategy/trend/positions.py index e040c77..237c77b 100644 --- a/py-client/strategy/trend/positions.py +++ b/py-client/strategy/trend/positions.py @@ -42,58 +42,61 @@ def manage_positions( remaining_cash = max(0.0, available) for position in positions: - code = position.stock_code - tick = ticks.get(code) - if code in runtime.account_cfg.excluded_codes: - log.info("[Position - ] 代码=%s，名称=%s，止盈=跳过，补仓=跳过，原因=已配置为排除股票", code, position.stock_name) - continue - if ( - not code - or position.open_price <= 0 - or position.volume <= 0 - or tick is None - or tick.last_price <= 0 - ): - log.warning("[Position - ] 代码=%s，名称=%s，止盈=跳过，补仓=跳过，原因=持仓或行情数据无效", code or "未知", position.stock_name) - continue + try: + code = position.stock_code + tick = ticks.get(code) + if code in runtime.account_cfg.excluded_codes: + log.info("[Position - ] 代码=%s，名称=%s，止盈=跳过，补仓=跳过，原因=已配置为排除股票", code, position.stock_name) + continue + if ( + not code + or position.open_price <= 0 + or position.volume <= 0 + or tick is None + or tick.last_price <= 0 + ): + log.warning("[Position - ] 代码=%s，名称=%s，止盈=跳过，补仓=跳过，原因=持仓或行情数据无效", code or "未知", position.stock_name) + continue - pnl_rate = round( - (tick.last_price - position.open_price) / position.open_price * 100, - 2, - ) - minimum_profit = calculate_min_profit_rate(position.open_price, 1) - profit_decision = handle_profit( - runtime=runtime, - position=position, - tick=tick, - pnl_rate=pnl_rate, - minimum_profit=minimum_profit, - ) - profit_action = profit_decision.message or "未触发" - loss_add_action = "未启用" - if runtime.account_cfg.enable_loss_add_position and market_ok: - loss_decision = handle_loss( + pnl_rate = round( + (tick.last_price - position.open_price) / position.open_price * 100, + 2, + ) + minimum_profit = calculate_min_profit_rate(position.open_price, 1) + profit_decision = handle_profit( runtime=runtime, position=position, tick=tick, pnl_rate=pnl_rate, - available=remaining_cash, + minimum_profit=minimum_profit, ) - remaining_cash -= loss_decision.reserved_cash - loss_add_action = loss_decision.message or "未触发" - elif runtime.account_cfg.enable_loss_add_position: - loss_add_action = "大盘信号不允许" + profit_action = profit_decision.message or "未触发" + loss_add_action = "未启用" + if runtime.account_cfg.enable_loss_add_position and market_ok: + loss_decision = handle_loss( + runtime=runtime, + position=position, + tick=tick, + pnl_rate=pnl_rate, + available=remaining_cash, + ) + remaining_cash -= loss_decision.reserved_cash + loss_add_action = loss_decision.message or "未触发" + elif runtime.account_cfg.enable_loss_add_position: + loss_add_action = "大盘信号不允许" - if pnl_rate>=0: - log.info( - "[Position ↑ ] 代码=%s，名称=%s，盈亏=%.2f%%，止盈=%s，补仓=%s", - code, position.stock_name, pnl_rate, profit_action, loss_add_action, - ) - else: - log.info( - "[Position ↓ ] 代码=%s，名称=%s，盈亏=%.2f%%，止盈=%s，补仓=%s", - code, position.stock_name, pnl_rate, profit_action, loss_add_action, - ) + if pnl_rate>=0: + log.info( + "[Position ↑ ] 代码=%s，名称=%s，盈亏=%.2f%%，止盈=%s，补仓=%s", + code, position.stock_name, pnl_rate, profit_action, loss_add_action, + ) + else: + log.info( + "[Position ↓ ] 代码=%s，名称=%s，盈亏=%.2f%%，止盈=%s，补仓=%s", + code, position.stock_name, pnl_rate, profit_action, loss_add_action, + ) + except Exception: + log.exception("[Position] 持仓处理异常，代码=%s，继续处理后续持仓", position.stock_code) def handle_profit( diff --git a/py-client/tests/test_ipo.py b/py-client/tests/test_ipo.py deleted file mode 100644 index 6593e35..0000000 --- a/py-client/tests/test_ipo.py +++ /dev/null @@ -1,140 +0,0 @@ -from __future__ import annotations - -import unittest -from datetime import datetime -from tempfile import TemporaryDirectory -from types import SimpleNamespace -from unittest.mock import patch - -from strategy.ipo.boot import AutoBuyIpo - - -RUN_TIME = datetime(2026, 8, 28, 10, 0) - - -class FakeClient: - def __init__(self, candidates=None, orders=None, deals=None, fail_codes=None): - self.candidates = candidates or {} - self.orders = orders or [] - self.deal_rows = deals or [] - self.fail_codes = set(fail_codes or []) - self.submissions = [] - self.closed = False - - def __enter__(self): - return self - - def __exit__(self, *_args): - self.closed = True - - def trading_dates(self, *_args): - return ["20260828"] - - def trade_detail_data(self, datatype): - self.assert_order_type = datatype - return self.orders - - def deals(self): - return self.deal_rows - - def ipo_data(self, ipo_type): - self.assert_ipo_type = ipo_type - return self.candidates - - def passorder(self, **kwargs): - code = kwargs["stock"] - self.submissions.append(kwargs) - if code in self.fail_codes: - raise RuntimeError("simulated rejection") - return {"status": "success", "order_ref": f"ref-{code}"} - - -class AutoBuyIpoTests(unittest.TestCase): - def _configs(self, directory, enabled=True): - return ( - SimpleNamespace( - qmt_base_url="http://qmt", - qmt_token="token", - qmt_data_dir=directory, - ), - SimpleNamespace(account_id="account-A", enable_auto_ipo=enabled), - ) - - def test_disabled_does_not_create_client(self): - with TemporaryDirectory() as directory: - global_cfg, account_cfg = self._configs(directory, enabled=False) - with ( - patch("strategy.ipo.boot.config.global_config", global_cfg), - patch("strategy.ipo.boot.config.account_config", account_cfg), - patch("strategy.ipo.boot.Client") as client_factory, - ): - self.assertEqual(AutoBuyIpo(RUN_TIME), 0) - client_factory.assert_not_called() - - def test_local_record_prevents_duplicate_after_restart(self): - candidates = { - "688001.SH": {"issuePrice": 10, "maxPurchaseNum": 1000}, - } - first = FakeClient(candidates=candidates) - second = FakeClient(candidates=candidates) - with TemporaryDirectory() as directory: - global_cfg, account_cfg = self._configs(directory) - with ( - patch("strategy.ipo.boot.config.global_config", global_cfg), - patch("strategy.ipo.boot.config.account_config", account_cfg), - patch("strategy.ipo.boot.Client", side_effect=[first, second]), - ): - self.assertEqual(AutoBuyIpo(RUN_TIME), 1) - self.assertEqual(AutoBuyIpo(RUN_TIME), 0) - - self.assertEqual(len(first.submissions), 1) - self.assertEqual(second.submissions, []) - self.assertTrue(first.closed) - self.assertTrue(second.closed) - - def test_broker_order_prevents_duplicate(self): - candidates = { - "688001.SH": {"issuePrice": 10, "maxPurchaseNum": 1000}, - } - client = FakeClient( - candidates=candidates, - orders=[{ - "m_strInstrumentID": "688001", - "m_strInsertDate": "20260828", - "m_strRemark": "IPO_SUBSCRIBE", - }], - ) - with TemporaryDirectory() as directory: - global_cfg, account_cfg = self._configs(directory) - with ( - patch("strategy.ipo.boot.config.global_config", global_cfg), - patch("strategy.ipo.boot.config.account_config", account_cfg), - patch("strategy.ipo.boot.Client", return_value=client), - ): - self.assertEqual(AutoBuyIpo(RUN_TIME), 0) - self.assertEqual(client.submissions, []) - - def test_one_rejection_does_not_stop_other_candidates(self): - candidates = { - "688001.SH": {"issuePrice": 10, "maxPurchaseNum": 1000}, - "688002.SH": {"issuePrice": 20, "maxPurchaseNum": 500}, - } - client = FakeClient(candidates=candidates, fail_codes={"688001.SH"}) - with TemporaryDirectory() as directory: - global_cfg, account_cfg = self._configs(directory) - with ( - patch("strategy.ipo.boot.config.global_config", global_cfg), - patch("strategy.ipo.boot.config.account_config", account_cfg), - patch("strategy.ipo.boot.Client", return_value=client), - ): - self.assertEqual(AutoBuyIpo(RUN_TIME), 1) - - self.assertEqual( - [item["stock"] for item in client.submissions], - ["688001.SH", "688002.SH"], - ) - self.assertTrue(client.closed) - - -if __name__ == "__main__": - unittest.main() diff --git a/py-client/tests/test_market.py b/py-client/tests/test_market.py deleted file mode 100644 index 244f2de..0000000 --- a/py-client/tests/test_market.py +++ /dev/null @@ -1,22 +0,0 @@ -from __future__ import annotations - -import unittest -from unittest.mock import patch - -from libs.market import market_allow_open, refresh_market - - -class MarketCacheTests(unittest.TestCase): - def test_refresh_updates_open_cache(self): - with patch("libs.market.get_json", return_value={"data": {"action": "UP"}}): - self.assertEqual(refresh_market("http://example"), "UP") - self.assertTrue(market_allow_open()) - - def test_refresh_failure_blocks_open(self): - with patch("libs.market.get_json", side_effect=OSError("offline")): - self.assertEqual(refresh_market("http://example"), "UNKNOWN") - self.assertFalse(market_allow_open()) - - -if __name__ == "__main__": - unittest.main() diff --git a/py-client/tests/test_signal.py b/py-client/tests/test_signal.py deleted file mode 100644 index acba6ad..0000000 --- a/py-client/tests/test_signal.py +++ /dev/null @@ -1,33 +0,0 @@ -from types import SimpleNamespace -import unittest -from unittest.mock import patch - -from libs.signal import SignalResult, fetch_signal, init_signals - - -class SignalTests(unittest.TestCase): - def test_fetch_failure_returns_empty_result(self): - with patch("libs.signal.get_json", side_effect=OSError("offline")): - self.assertEqual(fetch_signal("http://example", "/signals"), SignalResult()) - - def test_init_signals_continues_after_fetch_failure(self): - config = SimpleNamespace( - api_host="http://example", - signals={ - "failed": SimpleNamespace(url="/failed"), - "working": SimpleNamespace(url="/working"), - }, - ) - responses = [ - SignalResult(), - SignalResult(data={"000001.SZ": SimpleNamespace(signal_key="")}), - ] - with patch("libs.signal.fetch_signal", side_effect=responses): - result = init_signals(config, ["failed", "working"]) - - self.assertEqual(len(result), 1) - self.assertEqual(result[0].signal_key, "working") - - -if __name__ == "__main__": - unittest.main() diff --git a/py-client/tests/test_trend.py b/py-client/tests/test_trend.py deleted file mode 100644 index d5a537f..0000000 --- a/py-client/tests/test_trend.py +++ /dev/null @@ -1,359 +0,0 @@ -from __future__ import annotations - -import unittest -from concurrent.futures import ThreadPoolExecutor -from datetime import datetime, timedelta -from tempfile import TemporaryDirectory -from types import SimpleNamespace -from unittest.mock import patch - -from libs.grid_take_profit import GridState, GridTrailingTracker -from sdk import APIError, Assets, OrderItem, Portfolio, PositionItem, Tick -from strategy.trend.order import OrderBook, PlaceOrderRequest -from strategy.trend.open import do_open -from strategy.trend.positions import LOSS_TIERS, handle_loss, manage_positions -from strategy.trend.boot import RunOnce -from strategy.trend.state import STATUS_OK, State, StateItem - - -class FakeClient: - def __init__(self): - self.orders = [] - - def passorder_latest_tagged(self, op, code, volume, strategy_name, order_id): - self.orders.append((op, code, volume, strategy_name, order_id)) - return {"status": "success", "order_ref": f"broker-{len(self.orders)}"} - - -class FakeOrderClient: - def __init__(self, orders): - self.orders = orders - self.canceled = [] - - def trade_detail_data(self, _datatype): - return self.orders - - def cancel_by_id(self, order_id): - self.canceled.append(order_id) - - -class FailedOrderClient: - def passorder_latest_tagged(self, *_args): - raise APIError(502, "QMT did not return a valid order reference") - - -class TrendTests(unittest.TestCase): - def test_trend_order_id_format(self): - self.assertRegex(OrderBook.new_order_id(), r"^trend-[0-9a-f]{24}$") - - def test_open_records_only_pending_order(self): - with TemporaryDirectory() as directory: - state = State.for_strategy(directory, "trend", "A") - forgotten = [] - runtime = SimpleNamespace( - client=FakeClient(), - orders=OrderBook(), - state=state, - open_watch=SimpleNamespace(forget=forgotten.append), - ) - - do_open(runtime, "000001.SZ", 100, "morning", 12.345) - - item = state.get("000001.SZ") - self.assertRegex(item.base_order_id, r"^trend-[0-9a-f]{24}$") - self.assertEqual(item.base_qty, 0) - self.assertEqual(item.base_cost, 0) - self.assertEqual(item.base_status, "ING") - self.assertEqual(forgotten, ["000001.SZ"]) - - def test_pending_base_order_survives_position_delay(self): - with TemporaryDirectory() as directory: - state = State.for_strategy(directory, "trend", "A") - state.set(StateItem( - "000001.SZ", - base_order_id="trend-12345678", - base_status="ING", - )) - pending = OrderItem( - "1", "000001.SZ", "BUY", "", "50", None, 100, - local_order_id="trend-12345678", - ) - - state.reconcile([], [pending]) - - self.assertEqual( - state.get("000001.SZ").base_order_id, - "trend-12345678", - ) - - def test_grid_states_and_account_isolation(self): - tracker = GridTrailingTracker(1) - self.assertEqual(tracker.observe("A:code", 2.1).state, GridState.ARMED) - self.assertEqual(tracker.observe("A:code", 3.1).state, GridState.RAISED) - self.assertEqual(tracker.observe("A:code", 2.9).state, GridState.RETREAT) - self.assertEqual(tracker.observe("B:code", 2.9).state, GridState.ARMED) - tracker.retain([]) - self.assertEqual(tracker.observe("A:code", 2.9).state, GridState.ARMED) - - def test_order_book_locks_duplicate_order(self): - client = FakeClient() - book = OrderBook() - request = PlaceOrderRequest(client, 23, "000001.SZ", 100, "local", "morning") - self.assertTrue(book.place(request)) - self.assertTrue(book.busy("000001.SZ", "BUY")) - - def test_order_api_error_returns_false_with_traceback(self): - book = OrderBook() - request = PlaceOrderRequest(FailedOrderClient(), 23, "000001.SZ", 100, "local", "morning") - - with self.assertLogs(level="ERROR") as captured: - self.assertFalse(book.place(request)) - - output = "\n".join(captured.output) - self.assertIn("HTTP状态=502", output) - self.assertIn("Traceback", output) - - def test_refresh_tracks_active_and_completed_and_cancels_expired(self): - old = datetime.now() - timedelta(seconds=20) - orders = [ - OrderItem("active", "A", "BUY", "", "49", old, 100), - OrderItem("completed", "B", "SELL", "", "56", old, 100), - OrderItem("canceled", "C", "BUY", "", "54", old, 100), - OrderItem("failed", "D", "BUY", "", "57", old, 100), - ] - client = FakeOrderClient(orders) - book = OrderBook(cancel_timeout_sec=10) - - book.refresh(client, orders) - - self.assertEqual({item.id for item in book.data}, {"completed"}) - self.assertEqual(client.canceled, ["active"]) - - def test_position_dataclasses_execute_without_type_error(self): - with TemporaryDirectory() as directory: - state = State.for_strategy(directory, "trend", "A") - position = PositionItem( - stock_code="000001.SZ", volume=100, can_use_volume=100, - open_price=10, market_value=1000, - ) - state.sync_positions([position]) - runtime = SimpleNamespace( - client=FakeClient(), state=state, orders=OrderBook(), - open_watch=SimpleNamespace(forget=lambda _code: None), - add_watch=SimpleNamespace(triggered=lambda *_args: False, forget=lambda _code: None), - profit_tracker=GridTrailingTracker(1), - account_cfg=SimpleNamespace( - account_id="A", excluded_codes=[], grid_step_pct=1, - enable_loss_add_position=False, buy_value=5000, - strategy="trend", - ), - ) - manage_positions(runtime, {"000001.SZ": Tick(last_price=10.1)}, [position], True, 5000) - - def test_position_log_contains_code_name_profit_and_loss_actions(self): - with TemporaryDirectory() as directory: - state = State.for_strategy(directory, "trend", "A") - position = PositionItem( - stock_code="000001.SZ", stock_name="平安银行", volume=100, - can_use_volume=100, open_price=10, market_value=1000, - ) - state.sync_positions([position]) - runtime = SimpleNamespace( - client=FakeClient(), state=state, orders=OrderBook(), - add_watch=SimpleNamespace(triggered=lambda *_args: False), - profit_tracker=GridTrailingTracker(1), - account_cfg=SimpleNamespace( - account_id="A", excluded_codes=[], grid_step_pct=1, - enable_loss_add_position=False, buy_value=5000, - strategy="trend", - ), - ) - - with self.assertLogs(level="INFO") as captured: - manage_positions( - runtime, - {"000001.SZ": Tick(last_price=10.1)}, - [position], - True, - 5000, - ) - - output = "\n".join(captured.output) - self.assertIn("代码=000001.SZ", output) - self.assertIn("名称=平安银行", output) - self.assertIn("止盈=未触发", output) - self.assertIn("补仓=未启用", output) - - def test_loss_tier_boundary_does_not_overflow(self): - self.assertEqual(len(LOSS_TIERS), 2) - with TemporaryDirectory() as directory: - state = State.for_strategy(directory, "trend", "A") - position = PositionItem(stock_code="A", volume=100, open_price=10, market_value=1000) - state.sync_positions([position]) - item = state.get("A") - item.added_num = len(LOSS_TIERS) - state.set(item) - runtime = SimpleNamespace( - state=state, account_cfg=SimpleNamespace(buy_value=5000, strategy="trend"), - add_watch=SimpleNamespace(triggered=lambda *_args: True), orders=OrderBook(), - client=FakeClient(), - ) - decision = handle_loss(runtime, position, Tick(last_price=5), -60, 5000) - self.assertFalse(decision.submitted) - - def test_loss_tiers_zero_and_one(self): - with TemporaryDirectory() as directory: - state = State.for_strategy(directory, "trend", "A") - position = PositionItem(stock_code="A", volume=100, open_price=10, market_value=1000) - state.sync_positions([position]) - runtime = SimpleNamespace( - state=state, account_cfg=SimpleNamespace(buy_value=5000, strategy="trend"), - add_watch=SimpleNamespace(triggered=lambda *_args: False), - orders=OrderBook(), client=FakeClient(), - ) - first = handle_loss(runtime, position, Tick(last_price=7), -30, 5000) - self.assertIn("等待", first.message) - item = state.get("A") - item.added_num = 1 - state.set(item) - before_second_tier = handle_loss(runtime, position, Tick(last_price=6), -40, 5000) - self.assertEqual(before_second_tier.message, "") - second = handle_loss(runtime, position, Tick(last_price=5), -50, 5000) - self.assertIn("等待", second.message) - - def test_reconcile_split_orders_complete_only_when_all_are_status_56(self): - with TemporaryDirectory() as directory: - state = State.for_strategy(directory, "trend", "A") - position = PositionItem(stock_code="A", volume=100, open_price=10) - state.set(StateItem("A", base_order_id="local-1", base_status="ING")) - completed = OrderItem( - "1", "A", "BUY", "", "56", None, 50, "local-1", - traded_volume=50, trade_price=10.1, - ) - processing = OrderItem("2", "A", "BUY", "", "50", None, 50, "local-1") - - state.reconcile([position], [completed, processing]) - self.assertEqual(state.get("A").base_status, "ING") - - state.reconcile( - [position], - [ - completed, - OrderItem( - "2", "A", "BUY", "", "56", None, 50, "local-1", - traded_volume=50, trade_price=10.3, - ), - ], - ) - item = state.get("A") - self.assertEqual(item.base_status, STATUS_OK) - self.assertEqual(item.base_qty, 100) - self.assertEqual(item.base_cost, 10.2) - - canceled = OrderItem("2", "A", "BUY", "", "54", None, 50, "local-1") - item = state.get("A") - item.base_status = "ING" - state.set(item) - state.reconcile([position], [completed, canceled]) - self.assertEqual(state.get("A").base_status, "") - - def test_reconcile_records_filled_add_order(self): - with TemporaryDirectory() as directory: - state = State.for_strategy(directory, "trend", "A") - position = PositionItem(stock_code="A", volume=200, open_price=10) - state.set(StateItem( - "A", - base_qty=100, - base_cost=10, - base_status=STATUS_OK, - added_order_id="add-1", - added_status="ING", - )) - completed = OrderItem( - "1", "A", "BUY", "", "56", None, 100, "add-1", - traded_volume=100, trade_amount=950, - ) - - state.reconcile([position], [completed]) - - item = state.get("A") - self.assertEqual(item.added_status, STATUS_OK) - self.assertEqual(item.added_num, 1) - self.assertEqual(item.added_qty, 100) - self.assertEqual(item.added_cost, 9.5) - - def test_low_cash_still_runs_position_management(self): - client = SimpleNamespace( - portfolio=lambda: Portfolio( - assets=Assets(total=10000, available=10), - positions={"A": PositionItem(stock_code="A", volume=100, open_price=10)}, - orders=[], - ), - full_tick=lambda _codes: {"A": Tick(last_price=11)}, - ) - with ThreadPoolExecutor(max_workers=2) as executor: - runtime = SimpleNamespace( - client=client, - account_cfg=SimpleNamespace(min_cash_ratio=0.1), - global_cfg=SimpleNamespace(api_host="http://example"), - orders=SimpleNamespace(refresh=lambda _client, _orders: None, data=[]), - state=SimpleNamespace( - codes=["A"], - reconcile=lambda *_args: None, - ), - executor=executor, - ) - with ( - patch("strategy.trend.boot.trading_time", return_value=True), - patch("strategy.trend.boot.market_allow_open", return_value=True), - patch("strategy.trend.boot.open_signal") as open_mock, - patch("strategy.trend.boot.manage_positions") as manage_mock, - ): - RunOnce(runtime, []) - open_mock.assert_not_called() - manage_mock.assert_called_once() - - def test_state_without_broker_order_allows_reopen(self): - with TemporaryDirectory() as directory: - state = State.for_strategy(directory, "trend", "A") - state.set(StateItem( - "A", - base_order_id="missing-order", - base_qty=100, - base_status="ING", - )) - state.save() - client = SimpleNamespace( - portfolio=lambda: Portfolio( - assets=Assets(total=10000, available=5000), - positions={}, - orders=[], - ), - full_tick=lambda _codes: {"A": Tick(last_price=10)}, - ) - signal = SimpleNamespace(code="A", signal_key="morning") - with ThreadPoolExecutor(max_workers=2) as executor: - runtime = SimpleNamespace( - client=client, - account_cfg=SimpleNamespace(min_cash_ratio=0.1), - global_cfg=SimpleNamespace(api_host="http://example"), - orders=OrderBook(), - state=state, - open_watch=SimpleNamespace(forget=lambda _code: None), - add_watch=SimpleNamespace(forget=lambda _code: None), - executor=executor, - ) - with ( - patch("strategy.trend.boot.trading_time", return_value=True), - patch("strategy.trend.boot.market_allow_open", return_value=True), - patch("strategy.trend.boot.open_signal") as open_mock, - patch("strategy.trend.boot.manage_positions"), - ): - RunOnce(runtime, [signal]) - - open_mock.assert_called_once() - self.assertEqual(state.codes, []) - - -if __name__ == "__main__": - unittest.main() diff --git a/py-client/tests/test_trend_busy.py b/py-client/tests/test_trend_busy.py deleted file mode 100644 index 2772509..0000000 --- a/py-client/tests/test_trend_busy.py +++ /dev/null @@ -1,42 +0,0 @@ -import unittest -from datetime import datetime, timedelta -from unittest.mock import Mock - -from sdk import OrderItem -from strategy.trend.order import OrderBook, PlaceOrderRequest - - -class TrendBusyTests(unittest.TestCase): - def test_snapshot_blocks_order_without_local_cache(self): - book = OrderBook() - client = Mock() - order = OrderItem("1", "A", "BUY", "", "50", None, 100) - book.refresh(client, [order]) - self.assertTrue(book.busy("A", "BUY")) - self.assertFalse(book.busy("A", "SELL")) - self.assertFalse(book.place(PlaceOrderRequest(client, 23, "A", 100, "local", "trend"))) - client.passorder.assert_not_called() - book.refresh(client, []) - self.assertFalse(book.busy("A", "BUY")) - - def test_cancel_request_keeps_order_busy_until_terminal_snapshot(self): - book = OrderBook() - client = Mock() - order = OrderItem("1", "A", "BUY", "", "50", datetime.now() - timedelta(seconds=20), 100) - book.refresh(client, [order]) - client.cancel_by_id.assert_called_once_with("1") - self.assertTrue(book.busy("A", "BUY")) - order.status = "54" - book.refresh(client, [order]) - self.assertFalse(book.busy("A", "BUY")) - - def test_empty_snapshot_keeps_local_cache_protection(self): - book = OrderBook() - client = Mock() - client.passorder.return_value = {"status": "success", "order_ref": "1"} - request = PlaceOrderRequest(client, 23, "A", 100, "local", "trend") - self.assertTrue(book.place(request)) - book.refresh(client, []) - self.assertTrue(book.busy("A", "BUY")) - self.assertFalse(book.place(request)) - client.passorder.assert_called_once() diff --git a/py-client/tests/test_zt.py b/py-client/tests/test_zt.py deleted file mode 100644 index f6018b0..0000000 --- a/py-client/tests/test_zt.py +++ /dev/null @@ -1,35 +0,0 @@ -from __future__ import annotations - -import unittest -from datetime import datetime -from tempfile import TemporaryDirectory - -from sdk import OrderItem, PositionItem -from strategy.zt.state import BUYING, DONE, SELLING, SOLD, TState, TStateItem - - -class ZTStateTests(unittest.TestCase): - def test_reconcile_marks_sell_and_buy_orders_completed(self): - with TemporaryDirectory() as directory: - state = TState.for_strategy(directory, "zt", "A") - state.set(TStateItem("A", 1000, 10, "2026-08-31", SELLING, "sell-1", 500, 11)) - position = PositionItem(stock_code="A", volume=500, open_price=10) - state.reconcile([position], [OrderItem("1", "A", "SELL", "", "56", datetime.now(), 500, "sell-1")], "2026-08-31") - self.assertEqual(state.get("A").phase, SOLD) - - item = state.get("A") - item.phase, item.buy_order_id = BUYING, "buy-1" - state.set(item) - state.reconcile([position], [OrderItem("2", "A", "BUY", "", "56", datetime.now(), 500, "buy-1")], "2026-08-31") - self.assertEqual(state.get("A").phase, DONE) - - def test_new_position_becomes_dcm_base_state(self): - with TemporaryDirectory() as directory: - state = TState.for_strategy(directory, "zt", "A") - state.reconcile([PositionItem(stock_code="A", volume=800, open_price=12.5)], [], "2026-08-31") - item = state.get("A") - self.assertEqual((item.base_qty, item.base_cost), (800, 12.5)) - - -if __name__ == "__main__": - unittest.main()