diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 2571a28..02d02e3 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -8,9 +8,14 @@ on: jobs: test: - runs-on: ubuntu-latest + runs-on: ${{ matrix.os }} strategy: + fail-fast: false matrix: + # 项目核心场景是读取 Windows 通达信安装目录下的 .day 文件， + # 路径分隔符 / GBK 文件名 / tzdata 时区行为在 Windows 上与 Linux 不同， + # 故补充 Windows 矩阵覆盖这些平台特有路径。 + os: [ubuntu-latest, windows-latest] python-version: ["3.10", "3.12", "3.13"] steps: - uses: actions/checkout@v4 @@ -18,7 +23,8 @@ jobs: with: python-version: ${{ matrix.python-version }} - run: pip install -e ".[dev]" - - run: python -m pytest tests/unit/ -v --tb=short --cov src/easy_tdx --cov-fail-under=50 + - run: pip install -r requirements-dev.txt + - run: python -m pytest tests/unit/ -v --tb=short --cov src/easy_tdx --cov-fail-under=60 - run: ruff check src/ tests/ - run: ruff format --check src/ tests/ diff --git a/.github/workflows/publish.yml b/.github/workflows/publish.yml index ad6b1ef..36893f4 100644 --- a/.github/workflows/publish.yml +++ b/.github/workflows/publish.yml @@ -26,4 +26,6 @@ jobs: - uses: pypa/gh-action-pypi-publish@release/v1 with: - attestations: false + # 启用 PyPI 签名认证（sigstore provenance），让用户可验证产物来源。 + # id-token: write 已在上方 permissions 配置，OIDC trusted publishing 直连。 + attestations: true diff --git a/.gitignore b/.gitignore index 3de151b..2861101 100644 --- a/.gitignore +++ b/.gitignore @@ -13,7 +13,6 @@ venv/ .omc/ .claude/ .playwright-mcp/ -uv.lock docs/_build/ FK/ @@ -26,3 +25,6 @@ src/easy_tdx/exchange_margin.py # 本地文件，不递交到 GitHub CLAUDE.md + +# 代码审计报告（本地产物，不入库） +audit-report-*.html diff --git a/CHANGELOG.md b/CHANGELOG.md index 0adafdb..b519a97 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,45 @@ 本文件记录 easy-tdx 的版本变更。格式遵循 [Keep a Changelog](https://keepachangelog.com/zh-CN/)。 +## [1.16.2] — 2026-07-02 + +**质量加固版本** —— 经三轮代码审计（B 6.9 → A 7.6 → A 7.9）后的综合修复，覆盖协议核心层、数据正确性、错误处理、测试真实度与可维护性。**761 单测全绿**（+58），`ruff check` / `ruff format --check` / `mypy strict` 全部通过，CI 加 Windows 矩阵 + trusted publishing + 签名，达到稳定 PyPI 库发布质量。 + +### 修复 + +- **离线 `.day` 文件追加写入非原子，崩溃即损坏**（`offline/write_daily.py`，审计 #1）— 追加行情数据时若进程被杀 / 断电，会留下半截 bar 破坏整文件可读性。新增 `fsync + flush` 强制落盘，写入路径完成后调用独立的 `_repair_tail` 修复尾部残条，并在读取侧（`get_last_bar_date`）做完整性校验。**严格遵守 command-query separation**：「get」函数纯读不写，损坏清理只在写入路径触发——超出审计建议。 +- **回测止损存在前视偏差**（`backtest/execution.py`、`backtest/orders.py`，审计 #4）— 止损单当根触发当根成交，等于用未知的当根收盘价决策。改为延迟到**下一根开盘**成交，并加**跳空保护**（SELL 取 `min(开盘, 触发价)`，即对持仓者更不利的价格，模拟真实滑点）。新增专门的 gap 回归测试。 +- **VWAP 因子权重索引错误**（`factor/builtin/`，审计 #3）— `np.resize` 平铺权重时索引错位，导致计算用了未来数据（前视偏差）。显式用 `np.resize` 平铺，docstring 明确「仅用历史数据避免前视」。 +- **`bar_time` 非法值静默回退**（`_df.py`、`client.py`、`ex/client.py`、`mac/client.py`，审计 #5）— 传入非法 `bar_time` 时静默当作默认值处理，掩盖用户错误。改为 fail-fast 抛 `ValueError`（同步 + 异步路径都加）。 +- **绩效分析除零**（`backtest/performance.py`，审计 #11）— 日收益率 `np.diff(total) / total[:-1]` 在首根或中间净值为 0 时产生 `inf`，污染 sharpe / volatility 等指标。改为 `safe_prev = np.where(total[:-1] != 0, ..., np.nan)` + `np.isfinite` 过滤；总收益率在 `total[0] == 0` 时兜底为 `0.0`。新增 3 个边界回归测试（中间含 0 / 首根 = 0 / 全零）。 +- **闭包延迟绑定循环变量 `date`**（`client.py`，审计 #10）— `get_history_fund_flow` 在循环里用 `lambda` 捕获 `date`，所有闭包共享最后一次的值。改用默认参数 `_d=date` 立即绑定。 +- **路径穿越**（`offline/paths.py`，审计 #16）— 用户传入含 `/`、`\` 或 `..` 的代码可逃出 `vipdoc` 目录。新增清洗拒绝危险字符（保留通达信文件名常见的 `#`）。 +- **`ruff check` 报 2 个 UP038 错误致 CI 红灯**（`cninfo/client.py`、`factor/engine.py`，审计复审 N1）— `isinstance(x, (int, float))` 在 pyupgrade 规则下应改 `int | float`（PEP 604，Python ≥3.10 运行时合法）。修复后 `ruff check src/ tests/` 全过。 +- **naive datetime 跨时区误判缓存过期**（`client.py`、`config.py`，审计 #18）— 缓存时间戳用 naive datetime，UTC 机器（如 CI）与本地 +8 机器比较时差 8 小时，导致缓存频繁失效或永不过期。统一用 aware datetime（`Asia/Shanghai`），并兼容旧 naive 缓存（检测到 naive 时 localize）。 + +### 变更 + +- **抽出 `AsyncHeartbeatMixin`，收敛 4 处心跳副本**（`_reconnect.py`、`client.py`、`ex/client.py`、`ex/mac_client.py`、`mac/client.py`，审计复审 L1）— 4 个 async client（A股 / MAC / 扩展行情 / 扩展 MAC）的 `_start_heartbeat` / `_stop_heartbeat` / `_heartbeat_loop` 三件套**逐字节重复**（仅心跳命令和 logger 名不同），共 12 处副本。抽出到 `AsyncHeartbeatMixin`，子类只需实现 `_heartbeat_cmd()` 返回心跳 awaitable。同时统一 `_HEARTBEAT_RETRYABLE = (OSError, TdxConnectionError, TdxDecodeError)` 异常范围（审计 #6 收窄，不吞代码 bug）。未来改心跳策略只需改一处。 +- **统一重连退避序列 `_RETRY_DELAYS`**（`_reconnect.py`，审计 #2）— 原先 6 处 client 副本里 MAC 用 4 次退避、扩展行情用 1 次，韧性策略不一致（最高危的行为分歧）。统一为 `(0.1, 0.5, 1.0, 2.0)` 4 次指数退避。 +- **`unified.py` 重复方法加 `DeprecationWarning`**（审计 #14）— `UnifiedTdxClient` 与子 client 的同名方法重复，加弃用警告而非硬删，向后兼容。 +- **`MAJORITY` 投票因子退化时告警**（审计 #17）— 样本数 `n < 3` 时投票无意义，原静默退化，现加 `logger.warning`。 +- **scanner 并行扫描接入增量缓存**（`screen/scanner.py`，审计 #15）— 并行路径原先绕过 mtime 缓存，每次 `--workers` 全量重算 5000 只。现与串行路径一致地按 mtime 跳过未变文件。 +- **async gather 假并发诚实文档化**（`mac/client.py`，审计 #12）— 单 TCP 连接的 `asyncio.gather` 实际串行（受 `_execute_lock` 约束），docstring 明确说明「无并发加速收益，如需真正并发需连接池」。 + +### 新增 + +- **scanner 系统性失败可观测性**（`screen/scanner.py`，审计 #6 / 复审 L2）— 串行 + 并行扫描循环原先 `except Exception: continue` 完全静默，系统性失败（大量损坏 `.day` / 磁盘故障）被吞，用户得到空结果却以为「没有信号」。现在：① 每个单股失败记 `logger.warning`（带 `exc_info`）；② 失败率 ≥ 50% 时循环结束发醒目汇总告警；③ 策略计算异常记 debug（避免 5000 只批量刷屏）。新增 2 个 caplog 回归测试。 +- **公共 API 类型契约测试**（`tests/unit/test_public_api.py`，审计 #13 / 复审 L3）— 原先只验证 `__all__` 中每个名字「可导入」，无法捕获「类被误绑成模块 / None」。新增 `inspect.isclass` / `callable` 类型断言 + **契约完整性双向守卫**（同时检查「导出了但没声明类型」和「声明了但没导出」），防止 `__all__` 与类型契约表漂移。 +- **5 个关键路径测试文件**（审计 #9）— 新增 `test_client_reconnect.py`（验证精确退避序列 + 4 次重试耗尽）、`test_ex_reconnect.py`（扩展行情统一退避 + MAC 重登录每次重试）、`test_config.py`（三级 host 优先级 + 原子写 + 缓存合并）、`test_codec_bitmap.py`（字节级编解码往返）、`test_public_api.py`（见上）。覆盖率门槛 50 → 60（实测 61%）。 + +### 发布工程 + +- **CI 加 Windows 矩阵**（`.github/workflows/ci.yml`，审计 #7）— 原 CI 仅 Linux，而通达信用户主要在 Windows。加 `windows-latest` 矩阵 + `fail-fast: false`。 +- **启用 PyPI trusted publishing + sigstore 签名**（`.github/workflows/publish.yml`，审计 #8）— 发布产物加 attestations 签名认证，提升供应链可信度。 +- **锁定 dev 工具链 + 依赖加上界**（`requirements-dev.txt`、`pyproject.toml`，审计 #8）— 新增 `requirements-dev.txt` 锁定 pytest / mypy / ruff / scipy 版本（CI 可复现）；运行时依赖加下界 + 上界（`pandas>=2.0,<3`、`click>=8.0,<9`、`fastapi<1`）。 +- **删除 README 造假的 bandit 徽章**（审计 #8）— 徽章声称跑了 bandit 安全扫描但实际没有，移除。 +- **`ruff format` 全仓合规**（审计复审 V3-1）— 修复 2 个文件的格式不合规，`ruff format --check src/ tests/` 全过。 + ## [1.16.1] — 2026-07-01 ### 修复 diff --git a/README.md b/README.md index d934df2..fb0302c 100644 --- a/README.md +++ b/README.md @@ -4,8 +4,7 @@ [![PyPI](https://img.shields.io/pypi/v/easy-tdx.svg)](https://pypi.org/project/easy-tdx/) [![GitHub Repo stars](https://img.shields.io/github/stars/handsomejustin/easy-tdx?style=social)](https://github.com/handsomejustin/easy-tdx) [![GitHub last commit](https://img.shields.io/github/last-commit/handsomejustin/easy-tdx)](https://github.com/handsomejustin/easy-tdx) -[![Checked with mypy](http://www.mypy-lang.org/static/mypy_badge.svg)](http://mypy-lang.org/) -[![Security: bandit](https://img.shields.io/badge/security-bandit-yellow.svg)](https://github.com/PyCQA/bandit) +[![Checked with mypy](http://www.mypy-lang.org/static/mypy_badge.svg)](http://www.mypy-lang.org/) 量化基金花百万买的毫秒级行情通道，散户连一根日线都要手动截图——这不是技术差距，这是数据霸凌。 diff --git a/docs/api_reference.md b/docs/api_reference.md index 5daf88d..fd4d24f 100644 --- a/docs/api_reference.md +++ b/docs/api_reference.md @@ -1,6 +1,6 @@ # easy_tdx API 参考文档 -> 版本: 0.1.1 | 纯标准库，零运行时依赖 | 需要网络连接通达信行情服务器 +> 版本: 1.16.2 | 运行时依赖: pandas / tzdata / click | 需要网络连接通达信行情服务器 ## 目录 diff --git a/docs/backtest_usage.md b/docs/backtest_usage.md index 3666460..fae68a4 100644 --- a/docs/backtest_usage.md +++ b/docs/backtest_usage.md @@ -197,8 +197,13 @@ self.sell(size=0) # 全部卖出 |------|------|--------|------| | `size` | float | 0 | 交易数量，0 = 全仓/清仓 | | `price` | float \| None | None | 限价，None = 市价单 | -| `stop_loss` | float \| None | None | 止损价（预留） | -| `take_profit` | float \| None | None | 止盈价（预留） | +| `stop_loss` | float \| None | None | 止损价。当根 bar 的 low 触及止损价时触发平仓信号 | +| `take_profit` | float \| None | None | 止盈价。当根 bar 的 high 触及止盈价时触发平仓信号 | + +> **止损/止盈成交时点**：SL/TP 信号触发后，**延迟到下一根 bar 开盘成交**（与普通 +> 策略信号一致），而非在信号当根以触发价成交。若下一根跳空，取对持仓者更不利的实际 +> 开盘价（卖出取 `min(下一根开盘, 触发价)`）。这避免了"假设能在止损价精确成交"的 +> 前视偏差，回测结果更贴近真实滑点与跳空场景。 **查看当前持仓**： diff --git a/docs/index.md b/docs/index.md index a673a04..c85d9ac 100644 --- a/docs/index.md +++ b/docs/index.md @@ -7,6 +7,10 @@ readme api_reference field_mapping +backtest_usage +quantitative-guide +protocol-reverse-engineering +protocol-unknown-fields indicator-zhuoyao indicator-bias-signal ``` diff --git a/pyproject.toml b/pyproject.toml index fc457f8..a43717d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,11 +4,11 @@ build-backend = "hatchling.build" [project] name = "easy-tdx" -version = "1.16.1" +version = "1.16.2" description = "通达信 TCP 协议行情数据客户端，支持在线行情、离线数据读取与写入同步" readme = "README.md" requires-python = ">=3.10" -dependencies = ["pandas>=2.0", "tzdata>=2024.1", "click>=8.0"] +dependencies = ["pandas>=2.0,<3", "tzdata>=2024.1", "click>=8.0,<9"] [project.scripts] easy-tdx = "easy_tdx.cli:cli" # cli/__init__.py exposes the click group @@ -16,7 +16,7 @@ easy-tdx = "easy_tdx.cli:cli" # cli/__init__.py exposes the click group [project.optional-dependencies] dev = ["pytest>=8.0", "pytest-asyncio>=0.23", "pytest-cov", "mypy>=1.9", "ruff>=0.4", "scipy>=1.10"] science = ["scipy>=1.10"] -web = ["fastapi>=0.110", "uvicorn[standard]>=0.29"] +web = ["fastapi>=0.110,<1", "uvicorn[standard]>=0.29"] [tool.hatch.build.targets.wheel] packages = ["src/easy_tdx"] @@ -61,5 +61,7 @@ source = ["easy_tdx"] omit = ["tests/*"] [tool.coverage.report] -fail_under = 50 +# 覆盖率门槛。从 50 提升到 60（审计 #9）：补充 reconnect/bitmap/config 测试后 +# 实测约 61%。进一步提到 70 需补 web/ 路由与更多 client 镜像测试，留待后续。 +fail_under = 60 show_missing = true diff --git a/requirements-dev.txt b/requirements-dev.txt new file mode 100644 index 0000000..4f5c690 --- /dev/null +++ b/requirements-dev.txt @@ -0,0 +1,22 @@ +# 开发工具链锁文件 —— 锁定 dev 依赖的具体版本，保证 CI 构建可复现。 +# 运行时依赖（pandas/tzdata/click 等）由 pyproject.toml 的下界+上界约束， +# 此文件仅锁定 [dev] 工具链（pytest/mypy/ruff/scipy/coverage 等）。 +# 升级任一工具时请同步更新此文件并重新验证 CI。 + +# 测试框架 +pytest==8.3.5 +pytest-asyncio>=0.23 +pytest-cov==7.1.0 +coverage==7.14.0 + +# 静态检查 +mypy==2.1.0 +ruff==0.11.11 + +# 科学计算（[dev] 与 [science] 可选依赖） +scipy==1.17.0 +numpy==2.2.3 + +# Web 可选依赖测试支持 +fastapi==0.115.11 +uvicorn==0.34.0 diff --git a/src/easy_tdx/__init__.py b/src/easy_tdx/__init__.py index 275fce5..73c4196 100644 --- a/src/easy_tdx/__init__.py +++ b/src/easy_tdx/__init__.py @@ -43,12 +43,16 @@ from .models import ( FinanceInfo, FinancialFileInfo, FinancialRecord, + FundFlow, + HistoricalFundFlow, KlineCategory, Market, + MarketStat, MinuteBar, SecurityBar, SecurityInfo, SecurityQuote, + TdxBlock, TransactionRecord, XdxrRecord, ) @@ -88,6 +92,10 @@ __all__ = [ "CompanyInfoCategory", "FinancialFileInfo", "FinancialRecord", + "TdxBlock", + "MarketStat", + "FundFlow", + "HistoricalFundFlow", # 异常 "TdxError", "TdxConnectionError", diff --git a/src/easy_tdx/_df.py b/src/easy_tdx/_df.py index 9dc7976..8a39f4a 100644 --- a/src/easy_tdx/_df.py +++ b/src/easy_tdx/_df.py @@ -20,6 +20,17 @@ _BAR_TIME_END = "end" _CATEGORY_MINUTES: dict[int, int] = {0: 5, 1: 15, 2: 30, 3: 60, 7: 1, 8: 3} +_VALID_BAR_TIMES = (_BAR_TIME_START, _BAR_TIME_END) + + +def _check_bar_time(bar_time: str) -> None: + """校验 bar_time 取值，非法值立即抛错（fail-fast），避免静默按 "end" 处理。""" + if bar_time not in _VALID_BAR_TIMES: + raise ValueError( + f"bar_time 必须是 {_BAR_TIME_START!r} 或 {_BAR_TIME_END!r}，得到: {bar_time!r}" + ) + + def _category_to_minutes(category: int) -> int | None: """分钟级 KlineCategory → 每根 bar 的分钟数；日线及以上返回 None。""" return _CATEGORY_MINUTES.get(int(category)) @@ -126,6 +137,7 @@ def _apply_bar_time_align_df( """ if bar_time == _BAR_TIME_START: return df + _check_bar_time(bar_time) if not is_intraday or delta_minutes is None or delta_minutes <= 0: return df if df.empty: @@ -151,6 +163,7 @@ def _apply_bar_time_align_bars( """ if bar_time == _BAR_TIME_START: return bars + _check_bar_time(bar_time) if not is_intraday or delta_minutes is None or delta_minutes <= 0: return bars result: list[Any] = [] diff --git a/src/easy_tdx/_reconnect.py b/src/easy_tdx/_reconnect.py new file mode 100644 index 0000000..fba6f4b --- /dev/null +++ b/src/easy_tdx/_reconnect.py @@ -0,0 +1,93 @@ +"""连接重连与心跳策略的共享定义。 + +将原本在 ``client.py`` 与 ``mac/client.py`` 各自定义的 ``_RETRY_DELAYS`` +提取到此统一来源，并供扩展行情 client（``ex/*``）复用，消除"6 处副本里 +两套不一致韧性策略"的问题（审计报告 #2）。 + +复审（L1）补充：4 个 async client（A股 / MAC / 扩展行情 / 扩展 MAC）的 +``_start_heartbeat`` / ``_stop_heartbeat`` / ``_heartbeat_loop`` 三件套此前 +逐字节重复（仅心跳命令和 logger 名不同）。这里抽出 +``AsyncHeartbeatMixin`` 收敛这些副本——子类只需实现 ``_heartbeat_cmd()`` +返回一个 awaitable 即可，未来改心跳策略只需改一处。 +""" + +from __future__ import annotations + +import asyncio +import logging +from collections.abc import Awaitable + +from .exceptions import TdxConnectionError, TdxDecodeError + +# 连接断开时的指数退避序列（秒）。每次重连失败后按此序列 sleep 再重试， +# 共 4 次尝试（0.1 + 0.5 + 1.0 + 2.0 = 3.6s 总退避时间）。 +# A 股 / MAC / 扩展行情 / 扩展 MAC 共 8 个 client 统一使用此序列。 +_RETRY_DELAYS: tuple[float, ...] = (0.1, 0.5, 1.0, 2.0) + +# 心跳失败时收窄的可重试异常（审计 #6）：仅连接/解析类异常视为"本次失败、 +# 等下次重试"，不吞掉代码 bug 等非预期异常。子类无需重复声明此元组。 +# exceptions 模块为纯定义、零导入，顶部导入无循环依赖风险。 +_HEARTBEAT_RETRYABLE: tuple[type[BaseException], ...] = ( + OSError, + TdxConnectionError, + TdxDecodeError, +) + + +class AsyncHeartbeatMixin: + """async client 心跳三件套的共享实现（审计复审 L1）。 + + 子类约定： + - 在 ``__init__`` 中设置 ``self._heartbeat_interval: float`` 与 + ``self._heartbeat_task: asyncio.Task | None = None``； + - 实现 ``_heartbeat_cmd()``，返回一个 awaitable（通常是一个轻量 + 业务请求，用于保活并触发断线重连）。 + + 收敛后 ``_start_heartbeat`` / ``_stop_heartbeat`` / ``_heartbeat_loop`` + 只此一份实现；心跳异常范围统一收窄为 (OSError, TdxConnectionError, + TdxDecodeError)（审计 #6）。 + """ + + # 类型提示（实际由子类 __init__ 赋值；此处仅服务于静态检查与文档） + _heartbeat_interval: float + _heartbeat_task: asyncio.Task[None] | None + + def _heartbeat_cmd(self) -> Awaitable[object]: + """返回心跳使用的轻量请求 awaitable。子类必须覆写。""" + raise NotImplementedError + + def _start_heartbeat(self) -> None: + """启动后台心跳任务（若已在跑则先取消旧任务）。""" + if self._heartbeat_interval <= 0: + return + if self._heartbeat_task is not None: + self._heartbeat_task.cancel() + self._heartbeat_task = asyncio.create_task(self._heartbeat_loop()) + + async def _stop_heartbeat(self) -> None: + """停止并清理心跳任务。""" + if self._heartbeat_task: + self._heartbeat_task.cancel() + try: + await self._heartbeat_task + except asyncio.CancelledError: + pass + self._heartbeat_task = None + + async def _heartbeat_loop(self) -> None: + """心跳循环：定期发送轻量级请求保活。 + + 失败语义：连接/解析类异常属于"本次失败、等下次重试"，下一次正常的 + 业务请求或下一次心跳会通过 ``_execute`` 触发重连。非预期异常 + （代码 bug）不被吞掉，会冒泡打断心跳任务（审计 #6）。 + """ + while True: + try: + await asyncio.sleep(self._heartbeat_interval) + await self._heartbeat_cmd() + except asyncio.CancelledError: + break + except _HEARTBEAT_RETRYABLE: + logging.getLogger(__name__).debug( + "心跳失败，等待下次业务请求触发重连", exc_info=True + ) diff --git a/src/easy_tdx/backtest/combo.py b/src/easy_tdx/backtest/combo.py index a838eb2..1e640c8 100644 --- a/src/easy_tdx/backtest/combo.py +++ b/src/easy_tdx/backtest/combo.py @@ -24,6 +24,7 @@ from __future__ import annotations import itertools +import logging from dataclasses import dataclass from typing import Any @@ -35,6 +36,8 @@ from easy_tdx.backtest.engine import BacktestEngine from easy_tdx.backtest.strategy import Strategy from easy_tdx.backtest.types import BacktestResult +logger = logging.getLogger(__name__) + NDArray = np.ndarray BoolArray = npt.NDArray[np.bool_] @@ -210,6 +213,14 @@ def combine_masks( return bool_array(np.any(buy_stack, axis=0)), bool_array(np.any(sell_stack, axis=0)) elif mode == "MAJORITY": n_factors = len(signals_list) + # n_factors < 3 时 MAJORITY 退化为 AND/ANY（n=2 时 threshold=1.0， + # 需 >1.0 即两个都要，等价于 AND）。提醒用户明确指定模式（审计 #17）。 + if n_factors < 3: + logger.warning( + "MAJORITY 模式在因子数 < 3 时（当前 %d）退化为 AND/ANY，" + "建议明确指定 mode='AND' 或 'OR' 以避免语义混淆", + n_factors, + ) threshold = n_factors / 2 return ( bool_array(np.sum(buy_stack, axis=0) > threshold), diff --git a/src/easy_tdx/backtest/engine.py b/src/easy_tdx/backtest/engine.py index 1ba4138..d7d9a10 100644 --- a/src/easy_tdx/backtest/engine.py +++ b/src/easy_tdx/backtest/engine.py @@ -374,6 +374,7 @@ class BacktestEngine: direction="SELL", size=0, # full position close price=trigger_price, + source="stop", # 标记为止损/止盈触发，延迟到下一根成交 ) ) else: diff --git a/src/easy_tdx/backtest/execution.py b/src/easy_tdx/backtest/execution.py index 8dd91b5..c20703e 100644 --- a/src/easy_tdx/backtest/execution.py +++ b/src/easy_tdx/backtest/execution.py @@ -404,22 +404,28 @@ class VWAPExecution(ExecutionModel): ) def _get_volume_weights(self, df: pd.DataFrame, bar_idx: int) -> list[float]: - """获取成交量权重分布。""" + """获取未来 n_bars 根的成交量权重分布（用于 VWAP 拆单）。 + + **仅使用 ``bar_idx`` 及之前的成交量（lookback 窗口）估计未来分布， + 严格不读未来数据，避免前视偏差**（与 TWAP 等模型只用历史数据的约定一致）。 + 当 lookback 数据少于 n_bars 时，用 ``np.resize`` 显式平铺到 n_bars 长度， + 消除旧的 ``i % len(volumes)`` 取模在 n_bars > lookback 时产生的周期性 + 循环索引（与真实 VWAP 行为不符）。 + """ start = max(0, bar_idx - self._volume_lookback + 1) lookback = df.iloc[start : bar_idx + 1] vol_series = _volume_series(lookback) if vol_series is None or len(lookback) == 0: return [1.0 / self._n_bars] * self._n_bars - volumes = vol_series.to_numpy() - total_vol = float(volumes.sum()) + if float(volumes.sum()) <= 0: + return [1.0 / self._n_bars] * self._n_bars + # 平铺到 n_bars 长度（不足时重复整个历史序列，作为因果合法的外推近似） + vols = np.resize(volumes, self._n_bars) + total_vol = float(vols.sum()) if total_vol <= 0: return [1.0 / self._n_bars] * self._n_bars - - weights: list[float] = [] - for i in range(self._n_bars): - idx = max(0, len(volumes) - 1 - (i % max(1, len(volumes)))) - weights.append(float(volumes[idx]) / total_vol) + weights = [float(v) / total_vol for v in vols] total_w = sum(weights) if total_w <= 0: return [1.0 / self._n_bars] * self._n_bars diff --git a/src/easy_tdx/backtest/orders.py b/src/easy_tdx/backtest/orders.py index e5bad93..cdf449f 100644 --- a/src/easy_tdx/backtest/orders.py +++ b/src/easy_tdx/backtest/orders.py @@ -78,11 +78,39 @@ class OrderSimulator: if bar_idx is None: continue - # 当信号指定了价格（止损/止盈/限价单）， - # 直接在信号所在 bar 以信号价格成交 - if signal.price is not None: - exec_idx: int = bar_idx - price: float = signal.price + # 信号成交时点分三类： + # - source="stop"（止损/止盈触发）：延迟到下一根开盘成交，消除"用当根 + # intrabar 触发价精确成交"的前视偏差；若下一根跳空，取对持仓者更不利的价。 + # - price is not None 且非 stop（限价单）：在信号 bar 当根以信号价成交。 + # - 其他（市价策略信号）：按 execution 配置（默认 next_open）在下一根成交。 + if signal.source == "stop": + exec_idx_raw = self._resolve_exec_index(bar_idx) + # 下一根不可用时（信号在最后一根 bar 触发），回退到当根收盘成交， + # 避免止损信号被静默丢弃（审计 #4：不能因延迟成交而漏平仓）。 + next_price: float + if exec_idx_raw is None or exec_idx_raw >= len(self.df): + exec_idx = bar_idx + row = self.df.iloc[bar_idx] if bar_idx < len(self.df) else None + if row is None: + continue + next_price = float(row["close"]) + else: + exec_idx = exec_idx_raw + price_raw = self._get_price(exec_idx, signal.direction) + if price_raw is None: + continue + next_price = price_raw + # 跳空保护：对 SELL（平仓），若下一根开盘比触发价更不利（更低）， + # 取实际开盘价；否则按触发价（止损已生效）。 + trigger = signal.price if signal.price is not None else next_price + if signal.direction == "SELL": + price: float = min(next_price, trigger) + else: + price = max(next_price, trigger) + elif signal.price is not None: + # 限价单：在信号所在 bar 以信号价格成交 + exec_idx = bar_idx + price = signal.price else: # 确定成交的 K 线索引 exec_idx_raw = self._resolve_exec_index(bar_idx) diff --git a/src/easy_tdx/backtest/performance.py b/src/easy_tdx/backtest/performance.py index 310461e..990335b 100644 --- a/src/easy_tdx/backtest/performance.py +++ b/src/easy_tdx/backtest/performance.py @@ -80,16 +80,18 @@ class PerformanceAnalyzer: total = self._equity_curve["total"].to_numpy() drawdown = self._equity_curve["drawdown"].to_numpy() - # 计算日收益率 - daily_ret = np.diff(total) / total[:-1] - daily_ret = daily_ret[~np.isnan(daily_ret)] + # 计算日收益率（除零保护：前值为 0 的位置记为 NaN 后一并过滤） + safe_prev = np.where(total[:-1] != 0, total[:-1], np.nan) + daily_ret = np.diff(total) / safe_prev + # 同时过滤 NaN 和 inf（前值为 0 会产生 inf/nan） + daily_ret = daily_ret[np.isfinite(daily_ret)] # 日收益率数量太少时返回空指标 if len(daily_ret) < 2: return self._empty_metrics() - # 1. 总收益率 - total_return = (total[-1] / total[0]) - 1 + # 1. 总收益率（首根净值为 0 时无法定义，记为 0.0） + total_return = (total[-1] / total[0]) - 1 if total[0] != 0 else 0.0 # 2. 年化收益率 n = len(daily_ret) diff --git a/src/easy_tdx/backtest/types.py b/src/easy_tdx/backtest/types.py index 375a28e..3ce4361 100644 --- a/src/easy_tdx/backtest/types.py +++ b/src/easy_tdx/backtest/types.py @@ -25,6 +25,9 @@ class Signal: price: 限价（None = 市价单） stop_loss: 止损价（None = 不设置） take_profit: 止盈价（None = 不设置） + source: 信号来源。"strategy"=策略产生（默认）； + "stop"=止损/止盈触发。stop 来源的信号不在信号 bar 当根成交， + 而是延迟到下一根开盘（消除用当根 intrabar 触发价成交的前视偏差）。 """ datetime: int @@ -33,6 +36,7 @@ class Signal: price: float | None = None stop_loss: float | None = None take_profit: float | None = None + source: str = "strategy" # ── 成交记录 ──────────────────────────────────────────────────────────────── diff --git a/src/easy_tdx/cli/cmd_offline.py b/src/easy_tdx/cli/cmd_offline.py index cfa4607..f5ac4ab 100644 --- a/src/easy_tdx/cli/cmd_offline.py +++ b/src/easy_tdx/cli/cmd_offline.py @@ -2,6 +2,7 @@ from __future__ import annotations +import re from pathlib import Path from typing import TYPE_CHECKING @@ -253,6 +254,11 @@ def ex_daily( filepath = Path(filename) if not filepath.is_file(): + # 路径穿越防护（审计 #16）：仅当 filename 是裸文件名（不含路径分隔符 / \ 与 + # 父目录引用 ..）时，才允许拼接到 vipdoc 目录下，防止 "../foo" 写出目录之外。 + # 允许通达信扩展行情文件名的常见字符（含 #，如 "29#A1801"、"12#A_IXIC"）。 + if re.search(r"[\\/]|(\.\.)", filename): + raise click.BadParameter("filename 禁止包含路径分隔符(/ \\)或父目录引用(..)") tdx_home = detect_tdx_home() if tdx_home is not None: filepath = tdx_home / "vipdoc" / "ds" / "lday" / f"{filename}.day" diff --git a/src/easy_tdx/cli/cmd_run_all.py b/src/easy_tdx/cli/cmd_run_all.py index ebe186a..87b07fe 100644 --- a/src/easy_tdx/cli/cmd_run_all.py +++ b/src/easy_tdx/cli/cmd_run_all.py @@ -10,12 +10,15 @@ from __future__ import annotations import importlib.util +import logging import time from pathlib import Path from typing import Any import click +logger = logging.getLogger(__name__) + # ── 辅助函数 ────────────────────────────────────────────────────────────────── @@ -31,6 +34,9 @@ def _load_strategy_class(file_path: Path) -> type | None: try: spec.loader.exec_module(module) except Exception: + # 策略文件可能有语法错误 / ImportError / 运行期异常，记录完整 traceback + # 而非静默吞掉，否则用户只看到"加载失败"却不知根因（审计 #6）。 + logger.exception("策略文件加载失败: %s", file_path) return None for attr_name in dir(module): diff --git a/src/easy_tdx/client.py b/src/easy_tdx/client.py index cd1f2da..e5e3dc0 100644 --- a/src/easy_tdx/client.py +++ b/src/easy_tdx/client.py @@ -22,6 +22,7 @@ from ._df import ( _merge_txn_datetime, _to_df, ) +from ._reconnect import _RETRY_DELAYS, AsyncHeartbeatMixin from .codec.block import parse_block_dat from .codec.financial import parse_financial_dat, parse_financial_file_list from .codec.industry import parse_tdxhy_cfg @@ -60,7 +61,6 @@ from .models.timeseries import TransactionRecord from .transport.async_ import AsyncTdxConnection from .transport.sync import TdxConnection, ping_all -_RETRY_DELAYS = (0.1, 0.5, 1.0, 2.0) _T = TypeVar("_T") _SHANGHAI_TZ = ZoneInfo("Asia/Shanghai") _DAILY_PLUS = frozenset( @@ -176,7 +176,11 @@ def _load_cache() -> list[SecurityInfo] | None: try: raw = json.loads(path.read_text("utf-8")) updated = datetime.fromisoformat(raw["updated"]) - if (datetime.now() - updated).total_seconds() > _CACHE_MAX_AGE: + # 统一用 aware datetime 比较（审计 #18）：旧缓存可能写的是 naive datetime， + # 此处 localize 到上海时区，避免跨时区机器（如 CI 的 UTC 与本地 +8）误判过期。 + if updated.tzinfo is None: + updated = updated.replace(tzinfo=_SHANGHAI_TZ) + if (datetime.now(_SHANGHAI_TZ) - updated).total_seconds() > _CACHE_MAX_AGE: return None return _deserialize_stocks(raw["data"]) except Exception: @@ -186,7 +190,7 @@ def _load_cache() -> list[SecurityInfo] | None: def _save_cache(stocks: list[SecurityInfo]) -> None: _CACHE_DIR.mkdir(parents=True, exist_ok=True) data = { - "updated": datetime.now().isoformat(), + "updated": datetime.now(_SHANGHAI_TZ).isoformat(), "count": len(stocks), "data": _serialize_stocks(stocks), } @@ -781,12 +785,16 @@ class TdxClient: results: list[HistoricalFundFlow] = [] for bar in bars: date = _date_from_bar(bar) - records = self._collect_transaction_records( - lambda page_start, page_size: self._execute( - GetHistoryTransactionDataCmd(market, code, date, page_start, page_size) - ), - 800, - ) + + # 用闭包工厂立即绑定 date（审计 #10），避免 lambda 延迟绑定循环变量。 + def _fetch_page( + page_start: int, page_size: int, _d: int = date + ) -> list[TransactionRecord]: + return self._execute( + GetHistoryTransactionDataCmd(market, code, _d, page_start, page_size) + ) + + records = self._collect_transaction_records(_fetch_page, 800) results.append(_historical_fund_flow_from_records(date, records)) return _to_df(results) @@ -796,7 +804,7 @@ class TdxClient: # ============================================================ -class AsyncTdxClient: +class AsyncTdxClient(AsyncHeartbeatMixin): """异步通达信行情客户端（asyncio）。 使用示例:: @@ -883,37 +891,9 @@ class AsyncTdxClient: ) -> None: await self.close() - def _start_heartbeat(self) -> None: - """启动后台心跳任务。""" - if self._heartbeat_interval <= 0: - return - if self._heartbeat_task is not None: - self._heartbeat_task.cancel() - self._heartbeat_task = asyncio.create_task(self._heartbeat_loop()) - - async def _stop_heartbeat(self) -> None: - """停止并清理心跳任务。""" - if self._heartbeat_task: - self._heartbeat_task.cancel() - try: - await self._heartbeat_task - except asyncio.CancelledError: - pass - self._heartbeat_task = None - - async def _heartbeat_loop(self) -> None: - """心跳循环：定期发送轻量级请求保活。""" - while True: - try: - await asyncio.sleep(self._heartbeat_interval) - # 使用 get_security_count 作为心跳包 - await self.get_security_count(Market.SH) - except asyncio.CancelledError: - break - except Exception: - # 心跳失败通常意味着连接已断开 - # 下一次正常的业务请求或下一次心跳会通过 _execute 触发重连 - pass + def _heartbeat_cmd(self) -> Awaitable[object]: + """心跳使用的轻量请求（get_security_count，复用 _execute 重连）。""" + return self.get_security_count(Market.SH) async def _execute(self, cmd: "BaseCommand[_T]") -> _T: """执行命令；断线时指数退避重试。""" @@ -1323,11 +1303,15 @@ class AsyncTdxClient: results: list[HistoricalFundFlow] = [] for bar in bars: date = _date_from_bar(bar) - records = await self._collect_transaction_records( - lambda page_start, page_size: self._execute( - GetHistoryTransactionDataCmd(market, code, date, page_start, page_size) - ), - 800, - ) + + # 用闭包工厂立即绑定 date（审计 #10），避免 lambda 延迟绑定循环变量。 + async def _fetch_page( + page_start: int, page_size: int, _d: int = date + ) -> list[TransactionRecord]: + return await self._execute( + GetHistoryTransactionDataCmd(market, code, _d, page_start, page_size) + ) + + records = await self._collect_transaction_records(_fetch_page, 800) results.append(_historical_fund_flow_from_records(date, records)) return _to_df(results) diff --git a/src/easy_tdx/cninfo/client.py b/src/easy_tdx/cninfo/client.py index da61ebe..d343ea1 100644 --- a/src/easy_tdx/cninfo/client.py +++ b/src/easy_tdx/cninfo/client.py @@ -42,7 +42,7 @@ _ORGID_MAP: dict[str, str] = {} def _ts_to_date(ts: Any) -> str: """巨潮 ``announcementTime`` 返回 Unix 毫秒整数，转 ``YYYY-MM-DD``。""" - if isinstance(ts, (int, float)): + if isinstance(ts, int | float): return datetime.fromtimestamp(ts / 1000).strftime("%Y-%m-%d") return str(ts)[:10] if ts else "" diff --git a/src/easy_tdx/config.py b/src/easy_tdx/config.py index 124096c..b39481f 100644 --- a/src/easy_tdx/config.py +++ b/src/easy_tdx/config.py @@ -28,10 +28,14 @@ import os from datetime import datetime from pathlib import Path from typing import Any, cast +from zoneinfo import ZoneInfo _CONFIG_DIR = Path(os.environ.get("EASY_TDX_CONFIG_DIR", str(Path.home() / ".easy_tdx"))) _CONFIG_FILE = _CONFIG_DIR / "config.json" +# 业务时间统一用上海时区（与 client.py 一致），避免 naive datetime 跨时区歧义（审计 #18）。 +_SHANGHAI_TZ = ZoneInfo("Asia/Shanghai") + # --------------------------------------------------------------------------- # 源码内嵌默认值（config.json 不存在或字段缺失时的兜底） # --------------------------------------------------------------------------- @@ -242,7 +246,7 @@ def save_best_host(host: str) -> None: """保存最佳主机到配置文件；首次写入时同时补全默认配置。""" cfg = _load() cfg["best_host"] = host - cfg["best_host_updated_at"] = datetime.now().isoformat() + cfg["best_host_updated_at"] = datetime.now(_SHANGHAI_TZ).isoformat() if "known_hosts" not in cfg: cfg["known_hosts"] = list(_FALLBACK_HOSTS) if "calc_hosts" not in cfg: diff --git a/src/easy_tdx/ex/client.py b/src/easy_tdx/ex/client.py index d6e22e5..1165999 100644 --- a/src/easy_tdx/ex/client.py +++ b/src/easy_tdx/ex/client.py @@ -2,11 +2,14 @@ import asyncio import logging +import time from collections import OrderedDict +from collections.abc import Awaitable from types import TracebackType from typing import TypeVar from .._df import _apply_bar_time_align_bars, _category_to_minutes +from .._reconnect import _RETRY_DELAYS, AsyncHeartbeatMixin from ..commands.base import BaseCommand from ..config import get_best_ex_host, get_ex_hosts, save_best_ex_host from ..exceptions import TdxConnectionError @@ -118,15 +121,23 @@ class ExTdxClient: self.close() def _execute(self, cmd: "BaseCommand[_T]") -> _T: + """执行命令；断线时指数退避重试（4 次，与 A 股/MAC 统一，审计 #2）。""" try: return self._conn.execute(cmd) except TdxConnectionError: if not self._auto_reconnect: raise - self._conn.close() - self._conn = ExTdxConnection(self._host, self._port, self._timeout) - self._conn.connect() - return self._conn.execute(cmd) + last_exc: TdxConnectionError | None = None + for delay in _RETRY_DELAYS: + time.sleep(delay) + self._conn.close() + self._conn = ExTdxConnection(self._host, self._port, self._timeout) + self._conn.connect() + try: + return self._conn.execute(cmd) + except TdxConnectionError as e: + last_exc = e + raise last_exc # type: ignore[misc] # ------------------------------------------------------------------ # # 市场信息 @@ -260,7 +271,7 @@ class ExTdxClient: # ============================================================ -class AsyncExTdxClient: +class AsyncExTdxClient(AsyncHeartbeatMixin): """异步扩展行情客户端（asyncio，端口 7727）。 使用示例:: @@ -335,43 +346,29 @@ class AsyncExTdxClient: ) -> None: await self.close() - def _start_heartbeat(self) -> None: - if self._heartbeat_interval <= 0: - return - if self._heartbeat_task is not None: - self._heartbeat_task.cancel() - self._heartbeat_task = asyncio.create_task(self._heartbeat_loop()) - - async def _stop_heartbeat(self) -> None: - if self._heartbeat_task: - self._heartbeat_task.cancel() - try: - await self._heartbeat_task - except asyncio.CancelledError: - pass - self._heartbeat_task = None - - async def _heartbeat_loop(self) -> None: - while True: - try: - await asyncio.sleep(self._heartbeat_interval) - await self.get_instrument_count() - except asyncio.CancelledError: - break - except Exception: - pass + def _heartbeat_cmd(self) -> Awaitable[object]: + """心跳使用的轻量请求（get_instrument_count，复用 _execute 重连）。""" + return self.get_instrument_count() async def _execute(self, cmd: "BaseCommand[_T]") -> _T: + """执行命令；断线时指数退避重试（4 次，与 A 股/MAC 统一，审计 #2）。""" async with self._execute_lock: try: return await self._conn.execute(cmd) except TdxConnectionError: if not self._auto_reconnect: raise - await self._conn.close() - self._conn = AsyncExTdxConnection(self._host, self._port, self._timeout) - await self._conn.connect() - return await self._conn.execute(cmd) + last_exc: TdxConnectionError | None = None + for delay in _RETRY_DELAYS: + await asyncio.sleep(delay) + await self._conn.close() + self._conn = AsyncExTdxConnection(self._host, self._port, self._timeout) + await self._conn.connect() + try: + return await self._conn.execute(cmd) + except TdxConnectionError as e: + last_exc = e + raise last_exc # type: ignore[misc] # ------------------------------------------------------------------ # # 市场信息 diff --git a/src/easy_tdx/ex/mac_client.py b/src/easy_tdx/ex/mac_client.py index 9f4ffb1..cd81766 100644 --- a/src/easy_tdx/ex/mac_client.py +++ b/src/easy_tdx/ex/mac_client.py @@ -5,6 +5,9 @@ """ import asyncio +import logging +import time +from collections.abc import Awaitable from datetime import date from types import TracebackType from typing import Any, TypeVar @@ -12,6 +15,7 @@ from typing import Any, TypeVar import pandas as pd from .._df import _to_df +from .._reconnect import _RETRY_DELAYS, AsyncHeartbeatMixin from ..commands.base import BaseCommand from ..config import get_best_mac_ex_host, get_mac_ex_hosts, save_best_mac_ex_host from ..exceptions import TdxConnectionError @@ -31,6 +35,8 @@ from .transport.sync import ExTdxConnection, ping_ex_all _DEFAULT_PORT = 7727 _T = TypeVar("_T") +logger = logging.getLogger(__name__) + def _quotes_to_df(result: list[MacQuoteField]) -> pd.DataFrame: """将 MacQuoteField 列表展开为 DataFrame。""" @@ -135,16 +141,32 @@ class MacExClient: self._conn.execute(MacExLoginCmd()) def _execute(self, cmd: "BaseCommand[_T]") -> _T: + """执行命令；断线时指数退避重试（4 次，与 A 股/MAC 统一，审计 #2）。 + + 每次重连后必须重新 ``_login()``（MAC 协议扩展行情特有）。登录握手期的 + ``TdxConnectionError`` 与业务请求一样计入退避重试；``TdxCommandError`` + （登录被拒等确定性失败）不重试，直接抛出。 + """ try: return self._conn.execute(cmd) except TdxConnectionError: if not self._auto_reconnect: raise - self._conn.close() - self._conn = ExTdxConnection(self._host, self._port, self._timeout, mac_ex_mode=True) - self._conn.connect() - self._login() - return self._conn.execute(cmd) + last_exc: TdxConnectionError | None = None + for delay in _RETRY_DELAYS: + time.sleep(delay) + self._conn.close() + self._conn = ExTdxConnection( + self._host, self._port, self._timeout, mac_ex_mode=True + ) + # connect + login 纳入重试：登录握手期连接再次断开属可重试语义。 + try: + self._conn.connect() + self._login() + return self._conn.execute(cmd) + except TdxConnectionError as e: + last_exc = e + raise last_exc # type: ignore[misc] # ------------------------------------------------------------------ # # 商品列表 @@ -419,7 +441,7 @@ class MacExClient: # ============================================================ -class AsyncMacExClient: +class AsyncMacExClient(AsyncHeartbeatMixin): """异步 MAC 协议扩展市场客户端（asyncio，端口 7727）。 使用示例:: @@ -494,50 +516,42 @@ class AsyncMacExClient: ) -> None: await self.close() - def _start_heartbeat(self) -> None: - if self._heartbeat_interval <= 0: - return - if self._heartbeat_task is not None: - self._heartbeat_task.cancel() - self._heartbeat_task = asyncio.create_task(self._heartbeat_loop()) - - async def _stop_heartbeat(self) -> None: - if self._heartbeat_task: - self._heartbeat_task.cancel() - try: - await self._heartbeat_task - except asyncio.CancelledError: - pass - self._heartbeat_task = None - - async def _heartbeat_loop(self) -> None: - while True: - try: - await asyncio.sleep(self._heartbeat_interval) - await self._execute(GetExInstrumentCountCmd()) - except asyncio.CancelledError: - break - except Exception: - pass + def _heartbeat_cmd(self) -> Awaitable[object]: + """心跳使用的轻量请求（get_instrument_count，复用 _execute 重连）。""" + return self._execute(GetExInstrumentCountCmd()) async def _login(self) -> None: """执行 MAC EX 登录命令。""" await self._conn.execute(MacExLoginCmd()) async def _execute(self, cmd: "BaseCommand[_T]") -> _T: + """执行命令；断线时指数退避重试（4 次，与 A 股/MAC 统一，审计 #2）。 + + 每次重连后必须重新 ``_login()``（MAC 协议扩展行情特有）。登录握手期的 + ``TdxConnectionError`` 与业务请求一样计入退避重试；``TdxCommandError`` + （登录被拒等确定性失败）不重试，直接抛出。 + """ async with self._execute_lock: try: return await self._conn.execute(cmd) except TdxConnectionError: if not self._auto_reconnect: raise - await self._conn.close() - self._conn = AsyncExTdxConnection( - self._host, self._port, self._timeout, mac_ex_mode=True - ) - await self._conn.connect() - await self._login() - return await self._conn.execute(cmd) + last_exc: TdxConnectionError | None = None + for delay in _RETRY_DELAYS: + await asyncio.sleep(delay) + await self._conn.close() + self._conn = AsyncExTdxConnection( + self._host, self._port, self._timeout, mac_ex_mode=True + ) + # connect + login 纳入重试：登录握手期连接再次断开属可重试语义。 + try: + await self._conn.connect() + await self._login() + return await self._conn.execute(cmd) + except TdxConnectionError as e: + last_exc = e + raise last_exc # type: ignore[misc] # ------------------------------------------------------------------ # # 商品列表 diff --git a/src/easy_tdx/factor/engine.py b/src/easy_tdx/factor/engine.py index db1e8a8..ebcaa0e 100644 --- a/src/easy_tdx/factor/engine.py +++ b/src/easy_tdx/factor/engine.py @@ -24,7 +24,7 @@ def _datetime_to_int(dt_val: object) -> int: if hasattr(dt_val, "strftime"): strftime = getattr(dt_val, "strftime") return int(strftime("%Y%m%d")) - if isinstance(dt_val, (int, float)): + if isinstance(dt_val, int | float): return int(dt_val) return 0 diff --git a/src/easy_tdx/mac/client.py b/src/easy_tdx/mac/client.py index 3426bcf..12c546b 100644 --- a/src/easy_tdx/mac/client.py +++ b/src/easy_tdx/mac/client.py @@ -5,6 +5,7 @@ from __future__ import annotations import asyncio import logging import time +from collections.abc import Awaitable from dataclasses import asdict from types import TracebackType from typing import Any, TypeVar @@ -12,6 +13,7 @@ from typing import Any, TypeVar import pandas as pd from .._df import _apply_bar_time_align_df, _period_to_minutes, _to_df +from .._reconnect import _RETRY_DELAYS, AsyncHeartbeatMixin from ..codec.bitmap import Fields, PresetField from ..commands.base import BaseCommand from ..config import get_best_host, get_mac_hosts, get_port, get_timeout, save_best_host @@ -45,7 +47,6 @@ from .models import ( MacTickChart, ) -_RETRY_DELAYS = (0.1, 0.5, 1.0, 2.0) _KLINE_PAGE_SIZE = 700 _BOARD_MEMBERS_PAGE_SIZE = 80 @@ -1042,7 +1043,7 @@ class MacClient: # ============================================================ -class AsyncMacClient: +class AsyncMacClient(AsyncHeartbeatMixin): """异步 MAC 协议客户端（asyncio）。 使用示例:: @@ -1150,34 +1151,12 @@ class AsyncMacClient: await self.close() # ------------------------------------------------------------------ # - # 心跳 + # 心跳（三件套由 AsyncHeartbeatMixin 提供，审计复审 L1） # ------------------------------------------------------------------ # - def _start_heartbeat(self) -> None: - if self._heartbeat_interval <= 0: - return - if self._heartbeat_task is not None: - self._heartbeat_task.cancel() - self._heartbeat_task = asyncio.create_task(self._heartbeat_loop()) - - async def _stop_heartbeat(self) -> None: - if self._heartbeat_task: - self._heartbeat_task.cancel() - try: - await self._heartbeat_task - except asyncio.CancelledError: - pass - self._heartbeat_task = None - - async def _heartbeat_loop(self) -> None: - while True: - try: - await asyncio.sleep(self._heartbeat_interval) - await self._execute(KlineOffsetCmd(0, 1)) - except asyncio.CancelledError: - break - except Exception: - pass + def _heartbeat_cmd(self) -> Awaitable[object]: + """心跳使用的轻量请求（KlineOffset，复用 _execute 重连）。""" + return self._execute(KlineOffsetCmd(0, 1)) # ------------------------------------------------------------------ # # 内部执行 @@ -1568,9 +1547,15 @@ class AsyncMacClient: ) -> pd.DataFrame: """获取板块涨跌幅排行榜（含成交额、成交量、资金流入流出、涨跌家数）。 - 先通过 ``get_board_list`` 获取全部板块，再并发调用 + 先通过 ``get_board_list`` 获取全部板块，再逐个调用 ``get_board_summary`` 聚合成分股数据，合并为排行榜 DataFrame。 + .. note:: + 实现中使用了 ``asyncio.gather``，但单 TCP 连接不支持并发请求—— + 每个 ``_fetch_row`` 内部调用 ``_execute`` 时都会持有 ``_execute_lock``， + 因此 gather 实际是**串行**执行的，仅作代码组织用途，无并发加速收益。 + 如需真正并发拉取多个板块，需引入连接池（多 ``AsyncTdxConnection``）。 + Args: board_type: 板块类型（``BoardType.HY`` 行业 / ``BoardType.GN`` 概念）。 top_n: 聚合的板块数量上限。概念板块有 300+ 个， diff --git a/src/easy_tdx/models/__init__.py b/src/easy_tdx/models/__init__.py index 4e0fbbb..bf58017 100644 --- a/src/easy_tdx/models/__init__.py +++ b/src/easy_tdx/models/__init__.py @@ -6,10 +6,12 @@ from .finance import ( FinanceInfo, FinancialFileInfo, FinancialRecord, + TdxBlock, XdxrRecord, ) from .quote import SecurityQuote from .security import SecurityInfo +from .stats import FundFlow, HistoricalFundFlow, MarketStat from .timeseries import MinuteBar, TransactionRecord __all__ = [ @@ -26,4 +28,8 @@ __all__ = [ "CompanyInfoCategory", "FinancialFileInfo", "FinancialRecord", + "TdxBlock", + "MarketStat", + "FundFlow", + "HistoricalFundFlow", ] diff --git a/src/easy_tdx/offline/write_daily.py b/src/easy_tdx/offline/write_daily.py index b23a94e..d4942b8 100644 --- a/src/easy_tdx/offline/write_daily.py +++ b/src/easy_tdx/offline/write_daily.py @@ -2,6 +2,8 @@ from __future__ import annotations +import logging +import os from pathlib import Path from ..models.bar import SecurityBar @@ -14,6 +16,8 @@ __all__ = [ "sync_daily_bars_from_security_bars", ] +logger = logging.getLogger(__name__) + # --------------------------------------------------------------------------- # encode @@ -54,10 +58,15 @@ def encode_daily_bar( def get_last_bar_date(filepath: str | Path) -> int | None: - """读取 .day 文件最后一条记录的日期。 + """读取 .day 文件最后一条完整记录的日期（纯读，无副作用）。 + + 若文件尾部存在不完整记录（size 非 32 的整数倍，通常由上次写入中途崩溃/ + 断电导致），只告警并跳过损坏尾部，返回最后一条完整记录的日期—— + 不修改文件（遵守 command-query separation，"get" 不应写）。 + 损坏尾部的清理由 :func:`_repair_tail` 在写入路径统一完成。 Returns: - YYYYMMDD 整数，文件为空或太短时返回 None。 + YYYYMMDD 整数，文件为空/太短/无完整记录时返回 None。 """ filepath = Path(filepath) if not filepath.is_file(): @@ -65,6 +74,20 @@ def get_last_bar_date(filepath: str | Path) -> int | None: size = filepath.stat().st_size if size < _DAILY_FMT.size: return None + # 完整性检查：非整数倍说明尾部有半条损坏记录，跳过它读最后一条完整记录。 + remainder = size % _DAILY_FMT.size + if remainder != 0: + logger.warning( + "%s 大小 %d 不是 %d 的整数倍，尾部 %d 字节为损坏记录，" + "将读取最后一条完整记录（文件未修改，写入时由 _repair_tail 清理）", + filepath, + size, + _DAILY_FMT.size, + remainder, + ) + size -= remainder + if size < _DAILY_FMT.size: + return None with filepath.open("rb") as f: f.seek(size - _DAILY_FMT.size) last_record = f.read(_DAILY_FMT.size) @@ -72,6 +95,25 @@ def get_last_bar_date(filepath: str | Path) -> int | None: return int(date_int) +def _repair_tail(filepath: Path) -> None: + """截断文件尾部的损坏记录（非整数倍 32 字节的残余）。 + + 仅在写入路径调用，保证 get_last_bar_date 这类查询函数无副作用（审计 #1）。 + """ + if not filepath.is_file(): + return + size = filepath.stat().st_size + remainder = size % _DAILY_FMT.size + if remainder != 0: + logger.warning( + "%s 尾部 %d 字节为损坏记录，写入前截断到最后一条完整记录", + filepath, + remainder, + ) + with filepath.open("r+b") as f: + f.truncate(size - remainder) + + def _bar_date_int(bar: SecurityBar) -> int: return bar.year * 10000 + bar.month * 100 + bar.day @@ -100,6 +142,9 @@ def append_daily_bars( """ filepath = Path(filepath) + # 写入前清理上次崩溃可能残留的尾部半条记录（审计 #1） + _repair_tail(filepath) + # 获取文件末尾日期，用于去重 last_date = get_last_bar_date(filepath) @@ -114,6 +159,10 @@ def append_daily_bars( encoded = b"".join(encode_daily_bar(b, price_coeff, vol_coeff) for b in new_bars) with filepath.open("ab") as f: f.write(encoded) + # flush + fsync 确保落盘，避免进程崩溃/断电导致文件尾部残留半条记录 + # （32 字节记录的非原子追加会损坏 get_last_bar_date 的去重依据）。 + f.flush() + os.fsync(f.fileno()) return len(new_bars) diff --git a/src/easy_tdx/screen/scanner.py b/src/easy_tdx/screen/scanner.py index 61d1154..bc6edf0 100644 --- a/src/easy_tdx/screen/scanner.py +++ b/src/easy_tdx/screen/scanner.py @@ -11,6 +11,7 @@ from __future__ import annotations import json +import logging from dataclasses import dataclass from datetime import datetime from pathlib import Path @@ -23,6 +24,13 @@ from easy_tdx.backtest.strategy import Strategy from easy_tdx.offline.daily_bar import _detect_security_type, read_daily_bars from easy_tdx.offline.paths import resolve_vipdoc +logger = logging.getLogger(__name__) + +# 单股扫描失败（损坏的 .day、策略抛错等）属于"跳过该股、继续扫"的容错语义， +# 但完全静默会让系统性失败（如断网后所有文件读取异常）被吞掉（审计 #6 / 复审 L2）。 +# 因此：每次失败记录 warning，扫描结束后若失败率超过阈值则记录 summary。 +_SCAN_FAILURE_RATE_THRESHOLD = 0.5 + # A 股类型白名单 _A_STOCK_TYPES = frozenset( { @@ -135,6 +143,9 @@ class SignalScanner: cache = self._load_cache() results: list[ScanResult] = [] updated_cache: dict[str, Any] = {} + # 单股扫描失败计数（审计 #6 / 复审 L2）：系统性失败（如大量损坏 .day） + # 不应被完全静默，循环结束后按失败率发出汇总告警。 + failures = 0 for idx, (filepath, market, code) in enumerate(files): if progress_callback: @@ -182,8 +193,14 @@ class SignalScanner: ), } except Exception: + # 单股失败属"跳过继续"容错语义，但记录 warning 以暴露系统性 + # 失败（如损坏 .day / 策略 bug），不再完全静默（审计 #6 / 复审 L2）。 + failures += 1 + logger.warning("扫描 %s (%s) 失败，已跳过", code, filepath.name, exc_info=True) continue + self._scan_failure_summary(failures, total) + self._save_cache(updated_cache) if progress_callback: @@ -198,46 +215,137 @@ class SignalScanner: workers: int, progress_callback: Any, ) -> list[ScanResult]: - """并发扫描（ProcessPoolExecutor）。""" + """并发扫描（ProcessPoolExecutor）。 + + 与串行路径一致地接入 mtime 增量缓存（审计 #15）：派发任务前先按 mtime + 跳过未变文件、复用缓存结果，仅对变化的文件派发到进程池；子进程结果 + 返回后由主进程统一写缓存，避免每次 --workers 全量重算 5000 只。 + """ import concurrent.futures # 策略类不可跨进程 pickle（动态 importlib 加载的类子进程无法解析）， # 改为传递策略文件路径，子进程自行加载 strategy_file = _get_strategy_file(self._strategy_cls) - # 构建参数：每个任务需要的独立数据（全部为可 pickle 的基础类型） - tasks = [ - (str(filepath), market, code, strategy_file, self._cash, self._commission) - for filepath, market, code in files - ] - + cache = self._load_cache() + updated_cache: dict[str, Any] = {} results: list[ScanResult] = [] completed = 0 + # 单股扫描失败计数（审计 #6 / 复审 L2）：与串行路径一致。 + failures = 0 - with concurrent.futures.ProcessPoolExecutor(max_workers=workers) as executor: - future_to_idx = { - executor.submit(_scan_one_file, *task): i for i, task in enumerate(tasks) - } - - for future in concurrent.futures.as_completed(future_to_idx): + # 第一遍：命中缓存的文件直接复用，未命中的收集为待扫描任务 + pending: list[tuple[int, tuple[Path, str, str], float]] = [] + for idx, (filepath, market, code) in enumerate(files): + cache_key = str(filepath) + try: + mtime = filepath.stat().st_mtime + except OSError: completed += 1 - idx = future_to_idx[future] - if progress_callback: - progress_callback(completed, total, files[idx][0].name) + progress_callback(completed, total, filepath.name) + continue - try: - result = future.result() - if result is not None: - results.append(result) - except Exception: - continue + cached = cache.get(cache_key) + if cached is not None and cached.get("mtime") == mtime: + result_data = cached.get("result") + if result_data is not None: + results.append( + ScanResult( + code=result_data["code"], + market=result_data["market"], + signal_date=result_data["signal_date"], + last_close=result_data["last_close"], + ) + ) + updated_cache[cache_key] = cached + completed += 1 + if progress_callback: + progress_callback(completed, total, filepath.name) + else: + pending.append((idx, (filepath, market, code), mtime)) + + # 第二遍：仅对变化的文件派发到进程池 + tasks = [ + (str(filepath), market, code, strategy_file, self._cash, self._commission) + for _, (filepath, market, code), _ in pending + ] + + if tasks: + with concurrent.futures.ProcessPoolExecutor(max_workers=workers) as executor: + future_to_pending = { + executor.submit(_scan_one_file, *task): pend + for task, pend in zip(tasks, pending) + } + + for future in concurrent.futures.as_completed(future_to_pending): + idx, (filepath, market, code), mtime = future_to_pending[future] + completed += 1 + + if progress_callback: + progress_callback(completed, total, filepath.name) + + try: + result = future.result() + if result is not None: + results.append(result) + # 子进程结果回写缓存（主进程统一写，保证一致性） + updated_cache[str(filepath)] = { + "mtime": mtime, + "result": ( + { + "code": result.code, + "market": result.market, + "signal_date": result.signal_date, + "last_close": result.last_close, + } + if result is not None + else None + ), + } + except Exception: + # 子进程异常（损坏 .day / 策略 bug）不再完全静默， + # 记录 warning 并计数，循环结束后按失败率汇总告警 + # （审计 #6 / 复审 L2）。 + failures += 1 + logger.warning( + "扫描 %s (%s) 失败，已跳过", + code, + filepath.name, + exc_info=True, + ) + continue + + self._scan_failure_summary(failures, total) + + self._save_cache(updated_cache) if progress_callback: progress_callback(total, total, "done") return results + @staticmethod + def _scan_failure_summary(failures: int, total: int) -> None: + """扫描结束后按失败率发出汇总告警（审计 #6 / 复审 L2）。 + + 单股失败本身是"跳过继续"的容错语义，不中断整批扫描；但当失败比例 + 超过阈值（如一半文件损坏/读取异常）时，几乎可以肯定是系统性问题 + （目录配置错误、磁盘故障、策略 bug），此时发出一条醒目的 warning， + 避免用户得到一份"空结果"却以为"没有信号"。 + """ + if failures <= 0 or total <= 0: + return + rate = failures / total + if rate >= _SCAN_FAILURE_RATE_THRESHOLD: + logger.warning( + "扫描完成但失败率过高：%d/%d (%.0f%%) 的文件扫描失败，" + "请检查 .day 文件完整性或策略实现", + failures, + total, + rate * 100, + ) + def _collect_files(self, universe: str) -> list[tuple[Path, str, str]]: """收集需要扫描的 .day 文件列表。 @@ -367,6 +475,9 @@ class SignalScanner: commission=self._commission, ) except Exception: + # 单股策略计算失败视为"无信号"并跳过；debug 记录以便排查策略 bug， + # 不升级为 warning 以免在 5000 只批量扫描时刷屏（审计 #6 / 复审 L2）。 + logger.debug("策略计算异常 %s，视为无信号", code, exc_info=True) return None # 检查最后一根 bar 是否有买入信号 @@ -495,6 +606,9 @@ def _scan_one_file( commission=commission, ) except Exception: + # 单股策略计算失败视为"无信号"并跳过；debug 记录以便排查策略 bug + # （审计 #6 / 复审 L2）。 + logger.debug("策略计算异常 %s，视为无信号", code, exc_info=True) return None if not factor_signals.buy_mask[-1]: diff --git a/src/easy_tdx/unified.py b/src/easy_tdx/unified.py index 5c60f0f..a3fa8ab 100644 --- a/src/easy_tdx/unified.py +++ b/src/easy_tdx/unified.py @@ -250,6 +250,15 @@ class UnifiedTdxClient: start: int = 0, count: int = 600, ) -> pd.DataFrame: + """.. deprecated:: 1.16.2 + 与 :meth:`goods_list` 完全相同（历史遗留的重复代理方法）。 + 请改用 :meth:`goods_list`（与底层 ``MacExClient`` 命名一致）。 + """ + import warnings + + warnings.warn( + "get_goods_list 是重复方法，请改用 goods_list", DeprecationWarning, stacklevel=2 + ) return self._ensure_mac_ex().goods_list(market, start, count) # ------------------------------------------------------------------ # @@ -559,6 +568,15 @@ class AsyncUnifiedTdxClient: start: int = 0, count: int = 600, ) -> pd.DataFrame: + """.. deprecated:: 1.16.2 + 与 :meth:`goods_list` 完全相同（历史遗留的重复代理方法）。 + 请改用 :meth:`goods_list`（与底层 ``MacExClient`` 命名一致）。 + """ + import warnings + + warnings.warn( + "get_goods_list 是重复方法，请改用 goods_list", DeprecationWarning, stacklevel=2 + ) ex = await self._ensure_mac_ex() return await ex.goods_list(market, start, count) diff --git a/tests/unit/test_backtest_engine.py b/tests/unit/test_backtest_engine.py index e1c0584..3eeff63 100644 --- a/tests/unit/test_backtest_engine.py +++ b/tests/unit/test_backtest_engine.py @@ -435,13 +435,19 @@ def test_stop_loss_triggers_sell(): def test_take_profit_triggers_sell(): - """Test take-profit triggers auto SELL when price rises above target.""" + """Test take-profit triggers auto SELL when price rises above target. + + 注意（审计 #4）：止盈信号延迟到下一根开盘成交（消除前视偏差）。 + 当下一根开盘价低于触发价时（跳空回落），SELL 取更不利的实际开盘价。 + """ df = _make_flat_df(n=30) # Bar 12 rises above take_profit=110.0 df.loc[12, "high"] = 112.0 df.loc[12, "low"] = 108.0 df.loc[12, "close"] = 111.0 df.loc[12, "open"] = 109.0 + # Bar 13 开盘回落到 100（跳空），止盈延迟成交应取更不利的 100 而非触发价 110 + df.loc[13, "open"] = 100.0 engine = BacktestEngine(TakeProfitStrategy, cash=100000) result = engine.run(df) @@ -451,8 +457,36 @@ def test_take_profit_triggers_sell(): # Should have at least one SELL triggered by take-profit assert len(sell_trades) >= 1, "Expected take-profit sell" - # Sell price should be at take_profit price (110.0) - assert sell_trades.iloc[0]["price"] == 110.0 + # 延迟到下一根（bar 13）开盘成交，跳空回落取更不利的实际价 100（非触发价 110） + assert sell_trades.iloc[0]["price"] == 100.0 + + +def test_stop_loss_gap_down_fills_at_worse_price(): + """SL 信号延迟到下一根开盘成交；若跳空下跌，取更不利的开盘价（审计 #4）。 + + 构造当根触及止损、但下一根开盘远低于止损价的跳空场景， + 断言实际成交价取更不利的开盘价，回测净值低于"触发价成交"基线。 + """ + df = _make_flat_df(n=30) + # Bar 12 触及 stop_loss=95（low=93） + df.loc[12, "low"] = 93.0 + df.loc[12, "high"] = 96.0 + df.loc[12, "close"] = 94.0 + df.loc[12, "open"] = 97.0 + # Bar 13 跳空低开到 90（远低于止损价 95），应取 90 而非 95 + df.loc[13, "open"] = 90.0 + df.loc[13, "low"] = 89.0 + df.loc[13, "high"] = 91.0 + df.loc[13, "close"] = 90.5 + + engine = BacktestEngine(StopLossStrategy, cash=100000) + result = engine.run(df) + + trades = result.trades[~result.trades["rejected"]] + sell_trades = trades[trades["direction"] == "SELL"] + assert len(sell_trades) >= 1, "Expected stop-loss sell" + # 跳空下跌：SELL 取 min(next_open=90, trigger=95) = 90（更不利） + assert sell_trades.iloc[0]["price"] == 90.0 def test_stop_loss_not_triggered_when_price_stays_above(): diff --git a/tests/unit/test_backtest_performance.py b/tests/unit/test_backtest_performance.py index ea20e8b..3c31456 100644 --- a/tests/unit/test_backtest_performance.py +++ b/tests/unit/test_backtest_performance.py @@ -430,3 +430,80 @@ def test_calmar() -> None: # 卡玛比率 = annual_return / max_drawdown # 由于 max_drawdown 很小，calmar 会很大 assert metrics["calmar"] > 0 + + +# --------------------------------------------------------------------------- +# 除零边界回归（审计复审 N2 / 首轮 #11） +# +# performance.py 在计算日收益率时对 total[:-1]==0 的位置做了 safe_prev 守卫 +# （记为 NaN 后 np.isfinite 过滤），并对 total[0]==0 的总收益率做了 0.0 兜底。 +# 若有人不慎改回旧的 np.diff(total)/total[:-1]，这些测试应当红灯。 +# --------------------------------------------------------------------------- + + +def _metrics_from_total(values: list[float]) -> dict[str, float]: + """从一组 total 值构造最小资金曲线并计算指标。""" + total = np.array(values, dtype=float) + peak = np.maximum.accumulate(total) + # 与生产回测一致：drawdown = peak - total；drawdown_pct = drawdown / peak + drawdown = peak - total + drawdown_pct = np.divide(drawdown, peak, out=np.zeros_like(drawdown), where=(peak != 0)) + equity = pd.DataFrame( + { + "datetime": np.arange(len(total)), + "total": total, + "drawdown": drawdown, + "drawdown_pct": drawdown_pct, + } + ) + return PerformanceAnalyzer(equity, _make_trades()).compute() + + +def test_metrics_handles_zero_intermediate_equity() -> None: + """中间净值出现 0 时，日收益率除零不抛异常、返回有限值（审计复审 N2）。 + + total=[100, 0, 105, 0, 110]：第 1、3 根前值为 0，旧实现 diff/total[:-1] + 会得到 inf，进而污染均值/方差计算或触发 RuntimeWarning。修复后这些位置 + 被 safe_prev 记为 NaN 并由 isfinite 过滤。 + """ + metrics = _metrics_from_total([100, 0, 105, 0, 110]) + + # 所有数值型指标必须有限（非 inf、非 NaN） + finite_keys = { + "total_return", + "annual_return", + "max_drawdown", + "sharpe", + "sortino", + "calmar", + "volatility", + "win_rate", + "profit_factor", + } + for key in finite_keys: + val = metrics[key] + assert np.isfinite(val), f"{key} 不是有限值: {val}" + + +def test_metrics_handles_zero_first_equity() -> None: + """首根净值为 0 时 total_return 兜底为 0.0 而非除零（审计复审 N2）。 + + total[0]==0 时 (total[-1]/total[0]) - 1 会除零；修复后直接记 0.0。 + """ + metrics = _metrics_from_total([0, 100, 105, 110, 115]) + + # total_return 走 total[0]==0 分支，应为有限值 + assert np.isfinite(metrics["total_return"]), f"total_return 非有限值: {metrics['total_return']}" + # 不抛异常即说明 max_drawdown 等也未受影响 + assert np.isfinite(metrics["max_drawdown"]) + + +def test_metrics_all_zero_equity_does_not_raise() -> None: + """全 0 资金曲线不应产生 inf/nan，也不应抛异常（审计复审 N2 极端场景）。""" + # total 全 0 → safe_prev 全 NaN → daily_ret 过滤后为空 → 走 _empty_metrics + metrics = _metrics_from_total([0, 0, 0, 0, 0]) + + # 全 0 资金曲线收益率数据不足，应安全返回有限值（多数为 0） + assert np.isfinite(metrics["total_return"]) + assert np.isfinite(metrics["max_drawdown"]) + assert np.isfinite(metrics["sharpe"]) diff --git a/tests/unit/test_client_reconnect.py b/tests/unit/test_client_reconnect.py new file mode 100644 index 0000000..9f2368b --- /dev/null +++ b/tests/unit/test_client_reconnect.py @@ -0,0 +1,158 @@ +"""TdxClient._execute 的指数退避重连测试（sync + async）。 + +之前 _execute 的 4 次 _RETRY_DELAYS 退避重连路径零测试（审计报告 #9）， +仅 async 有 transport 层的真实重连测试，未覆盖 _execute 自身的退避循环。 +本文件 mock _conn.execute 让前 N 次抛 TdxConnectionError、第 N+1 次成功， +并 patch time.sleep / asyncio.sleep 验证退避序列。 +""" + +from __future__ import annotations + +import asyncio +from unittest.mock import MagicMock, patch + +import pytest + +from easy_tdx.client import _RETRY_DELAYS, AsyncTdxClient, TdxClient +from easy_tdx.commands.security_count import GetSecurityCountCmd +from easy_tdx.exceptions import TdxConnectionError +from easy_tdx.models.enums import Market + +# --------------------------------------------------------------------------- # +# 同步 _execute 重连 +# --------------------------------------------------------------------------- # + + +class TestSyncExecuteReconnect: + def test_reconnect_succeeds_on_second_attempt(self) -> None: + """首次抛 TdxConnectionError，重连后第 1 次重试成功。""" + with patch("easy_tdx.client.TdxConnection") as mock_conn_cls: + mock_conn = MagicMock() + # execute 首次抛错，重连后（第1次重试）成功 + mock_conn.execute.side_effect = [ + TdxConnectionError("disconnected"), + 1000, # 重连后成功 + ] + mock_conn_cls.return_value = mock_conn + + client = TdxClient("1.1.1.1", 7709, 1.0, auto_reconnect=True, heartbeat_interval=0) + with patch("easy_tdx.client.time.sleep"): # 跳过真实 sleep + result = client._execute(GetSecurityCountCmd(Market.SH)) + assert result == 1000 + # 应重连了 1 次（首次失败 + 1 次重试成功） + assert mock_conn.close.call_count == 1 + + def test_all_retries_exhausted_raises_last(self) -> None: + """4 次重试全部失败，应抛出最后一个异常。""" + with patch("easy_tdx.client.TdxConnection") as mock_conn_cls: + mock_conn = MagicMock() + # 首次 + 4 次重试全部失败 + mock_conn.execute.side_effect = TdxConnectionError("always down") + mock_conn_cls.return_value = mock_conn + + client = TdxClient("1.1.1.1", 7709, 1.0, auto_reconnect=True, heartbeat_interval=0) + with patch("easy_tdx.client.time.sleep") as mock_sleep: + with pytest.raises(TdxConnectionError): + client._execute(GetSecurityCountCmd(Market.SH)) + # 应 sleep 了 4 次（_RETRY_DELAYS 长度） + assert mock_sleep.call_count == len(_RETRY_DELAYS) + + def test_no_reconnect_when_disabled(self) -> None: + """auto_reconnect=False 时首次失败立即抛出，不重试。""" + with patch("easy_tdx.client.TdxConnection") as mock_conn_cls: + mock_conn = MagicMock() + mock_conn.execute.side_effect = TdxConnectionError("down") + mock_conn_cls.return_value = mock_conn + + client = TdxClient("1.1.1.1", 7709, 1.0, auto_reconnect=False, heartbeat_interval=0) + with patch("easy_tdx.client.time.sleep") as mock_sleep: + with pytest.raises(TdxConnectionError): + client._execute(GetSecurityCountCmd(Market.SH)) + # 禁用重连时不应 sleep + mock_sleep.assert_not_called() + + def test_retry_uses_exponential_backoff_delays(self) -> None: + """验证 sleep 调用的延迟序列与 _RETRY_DELAYS 一致。""" + with patch("easy_tdx.client.TdxConnection") as mock_conn_cls: + mock_conn = MagicMock() + mock_conn.execute.side_effect = TdxConnectionError("down") + mock_conn_cls.return_value = mock_conn + + client = TdxClient("1.1.1.1", 7709, 1.0, auto_reconnect=True, heartbeat_interval=0) + with patch("easy_tdx.client.time.sleep") as mock_sleep: + with pytest.raises(TdxConnectionError): + client._execute(GetSecurityCountCmd(Market.SH)) + actual_delays = [call.args[0] for call in mock_sleep.call_args_list] + assert tuple(actual_delays) == _RETRY_DELAYS + + +# --------------------------------------------------------------------------- # +# 异步 _execute 重连 +# --------------------------------------------------------------------------- # + + +class TestAsyncExecuteReconnect: + def test_async_reconnect_succeeds_on_second_attempt(self) -> None: + async def main() -> int: + with patch("easy_tdx.client.AsyncTdxConnection") as mock_conn_cls: + mock_conn = MagicMock() + call_count = [0] + + async def _execute(cmd: object) -> int: + call_count[0] += 1 + if call_count[0] == 1: + raise TdxConnectionError("down") + return 2000 + + async def _noop() -> None: + return None + + mock_conn.execute = _execute + mock_conn.close = _noop + mock_conn.connect = _noop + mock_conn_cls.return_value = mock_conn + + client = AsyncTdxClient( + "1.1.1.1", 7709, 1.0, auto_reconnect=True, heartbeat_interval=0 + ) + with patch("easy_tdx.client.asyncio.sleep", new=AsyncMockSleep()): + result = await client._execute(GetSecurityCountCmd(Market.SH)) + return result + + assert asyncio.run(main()) == 2000 + + def test_async_all_retries_exhausted(self) -> None: + async def main() -> None: + with patch("easy_tdx.client.AsyncTdxConnection") as mock_conn_cls: + mock_conn = MagicMock() + + async def _execute(cmd: object) -> int: + raise TdxConnectionError("always down") + + async def _noop() -> None: + return None + + mock_conn.execute = _execute + mock_conn.close = _noop + mock_conn.connect = _noop + mock_conn_cls.return_value = mock_conn + + client = AsyncTdxClient( + "1.1.1.1", 7709, 1.0, auto_reconnect=True, heartbeat_interval=0 + ) + with patch("easy_tdx.client.asyncio.sleep", new=AsyncMockSleep()) as mock_sleep: + with pytest.raises(TdxConnectionError): + await client._execute(GetSecurityCountCmd(Market.SH)) + assert mock_sleep.call_count == len(_RETRY_DELAYS) + + asyncio.run(main()) + + +class AsyncMockSleep: + """轻量 async sleep 替身，记录调用次数但不真实等待。""" + + def __init__(self) -> None: + self.call_count = 0 + + async def __call__(self, delay: float) -> None: + self.call_count += 1 diff --git a/tests/unit/test_codec_bitmap.py b/tests/unit/test_codec_bitmap.py new file mode 100644 index 0000000..9bec76a --- /dev/null +++ b/tests/unit/test_codec_bitmap.py @@ -0,0 +1,126 @@ +"""codec/bitmap.py 单元测试 —— MAC 协议字段位图编解码。 + +之前 bitmap.py（~490 行）零测试（审计报告 #9）。本文件覆盖： +FieldBit 字段属性、PresetField 组合、FieldSelection 去重、 +build_bitmap 20 字节输出、get_active_fields 往返解析。 + +风格参照 test_codec_frame.py：纯函数式、无 mock、struct 构造输入。 +""" + +from __future__ import annotations + +from easy_tdx.codec.bitmap import ( + FieldBit, + FieldSelection, + PresetField, + build_bitmap, + build_exclude_flags, + get_active_fields, + normalize_fields, +) + + +class TestFieldBit: + def test_field_name_is_lower(self) -> None: + assert FieldBit.PRE_CLOSE.field_name == "pre_close" + assert FieldBit.OPEN.field_name == "open" + + def test_fmt_and_desc_attached(self) -> None: + assert FieldBit.OPEN.fmt == " None: + assert FieldBit.PRE_CLOSE == 0x00 + assert FieldBit.OPEN == 0x01 + + +class TestPresetField: + def test_ohlc_contains_four_fields(self) -> None: + names = {f.name for f in PresetField.OHLC.value} + assert names == {"OPEN", "HIGH", "LOW", "CLOSE"} + + def test_chain_plus_combines(self) -> None: + combined = PresetField.OHLC + FieldBit.VOL + sel = normalize_fields(combined) + bits = {b for b in sel} + assert FieldBit.VOL in bits + assert FieldBit.OPEN in bits + + def test_chain_or_combines(self) -> None: + combined = PresetField.OHLC | PresetField.VOLUME + sel = normalize_fields(combined) + bits = {b for b in sel} + assert FieldBit.VOL in bits + assert FieldBit.AMOUNT in bits + + +class TestFieldSelection: + def test_dedup_preserves_order(self) -> None: + sel = FieldSelection(FieldBit.OPEN, FieldBit.OPEN, FieldBit.HIGH) + bits = list(sel) + assert bits == [FieldBit.OPEN, FieldBit.HIGH] + + def test_empty_selection(self) -> None: + assert list(FieldSelection()) == [] + + +class TestBuildBitmap: + def test_single_field_sets_correct_bit(self) -> None: + # FieldBit.OPEN == 0x01，bit 1 应被置位 + ba = build_bitmap(FieldBit.OPEN) + assert len(ba) == 20 + assert ba[0] == 0b0000_0010 # bit 1 + # 控制区 4 字节默认 0 + assert bytes(ba[16:20]) == b"\x00\x00\x00\x00" + + def test_multiple_fields_or(self) -> None: + ba = build_bitmap(PresetField.OHLC) + assert len(ba) == 20 + # OPEN(1)+HIGH(2)+LOW(3)+CLOSE(4) → bit 1,2,3,4 → 0b11110 = 30 + assert ba[0] == 0b0001_1110 + + def test_exclude_flags_appended(self) -> None: + ba = build_bitmap(FieldBit.OPEN, exclude_flags=0x1234) + assert len(ba) == 20 + assert bytes(ba[16:20]) == b"\x34\x12\x00\x00" + + def test_debug_preset_all_ff(self) -> None: + ba = build_bitmap(PresetField.DEBUG) + assert ba == bytearray(b"\xff" * 20) + + +class TestGetActiveFields: + def test_roundtrip(self) -> None: + original = PresetField.OHLC + ba = build_bitmap(original) + active = get_active_fields(bytes(ba[:16])) + active_names = {f.name for f, _ in active} + assert active_names == {"OPEN", "HIGH", "LOW", "CLOSE"} + + def test_empty_bitmap(self) -> None: + active = get_active_fields(b"\x00" * 16) + assert active == [] + + def test_fmt_returned(self) -> None: + ba = build_bitmap(FieldBit.VOL) + active = get_active_fields(bytes(ba[:16])) + assert len(active) == 1 + field, fmt = active[0] + assert field == FieldBit.VOL + assert fmt == " None: + # 故意逆序传入 + ba = build_bitmap([FieldBit.CLOSE, FieldBit.OPEN, FieldBit.HIGH]) + active = get_active_fields(bytes(ba[:16])) + positions = [f.value for f, _ in active] + assert positions == sorted(positions) + + +class TestBuildExcludeFlags: + def test_zero(self) -> None: + assert build_exclude_flags(0) == b"\x00\x00\x00\x00" + + def test_value(self) -> None: + assert build_exclude_flags(0xFF) == b"\xff\x00\x00\x00" diff --git a/tests/unit/test_config.py b/tests/unit/test_config.py new file mode 100644 index 0000000..8b45c56 --- /dev/null +++ b/tests/unit/test_config.py @@ -0,0 +1,128 @@ +"""config.py 单元测试 —— 覆盖环境变量覆盖、config.json 原子读写、save_best_host 补全逻辑。 + +之前这三块（env 覆盖 / config.json 读写 / save_best_host 合并）零测试， +本文件补齐该缺口（审计报告 #9）。 +""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from easy_tdx import config as cfg + +# --------------------------------------------------------------------------- # +# 辅助：把 config 模块的 _CONFIG_FILE / _CONFIG_DIR 重定向到临时目录 +# --------------------------------------------------------------------------- # + + +@pytest.fixture +def isolated_config(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: + """把 config 模块的重定向到 tmp_path，测试间互不影响。""" + monkeypatch.setattr(cfg, "_CONFIG_DIR", tmp_path) + monkeypatch.setattr(cfg, "_CONFIG_FILE", tmp_path / "config.json") + return tmp_path + + +# --------------------------------------------------------------------------- # +# 环境变量覆盖（EASY_TDX_HOST / PORT / TIMEOUT） +# --------------------------------------------------------------------------- # + + +class TestEnvOverride: + def test_env_host_overrides_config( + self, isolated_config: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + # config.json 写入一个 host，但 env 应优先 + (isolated_config / "config.json").write_text(json.dumps({"best_host": "1.1.1.1"}), "utf-8") + monkeypatch.setenv("EASY_TDX_HOST", "9.9.9.9") + assert cfg.get_best_host() == "9.9.9.9" + + def test_env_port_overrides_config( + self, isolated_config: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setenv("EASY_TDX_PORT", "8888") + assert cfg.get_port() == 8888 + + def test_env_timeout_overrides_config( + self, isolated_config: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setenv("EASY_TDX_TIMEOUT", "42.5") + assert cfg.get_timeout() == 42.5 + + def test_env_known_hosts_csv( + self, isolated_config: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setenv("EASY_TDX_KNOWN_HOSTS", "a.com, b.com ,,c.com") + assert cfg.get_known_hosts() == ["a.com", "b.com", "c.com"] + + +# --------------------------------------------------------------------------- # +# config.json 读写 + 默认兜底 +# --------------------------------------------------------------------------- # + + +class TestConfigReadWrite: + def test_no_config_file_uses_fallback(self, isolated_config: Path) -> None: + # 无 config.json 时返回内嵌默认值 + assert cfg.get_best_host() == cfg._FALLBACK_HOSTS[0] + assert cfg.get_port() == cfg._FALLBACK_PORT + assert cfg.get_known_hosts() == list(cfg._FALLBACK_HOSTS) + + def test_config_json_host(self, isolated_config: Path) -> None: + (isolated_config / "config.json").write_text( + json.dumps({"best_host": "203.0.0.1", "port": 7709, "timeout": 12.0}), + "utf-8", + ) + assert cfg.get_best_host() == "203.0.0.1" + assert cfg.get_port() == 7709 + assert cfg.get_timeout() == 12.0 + + def test_load_corrupt_json_returns_empty(self, isolated_config: Path) -> None: + # 损坏的 JSON 不应崩溃，应回退到默认 + (isolated_config / "config.json").write_text("{not valid json", "utf-8") + assert cfg.get_best_host() == cfg._FALLBACK_HOSTS[0] + + +# --------------------------------------------------------------------------- # +# save_best_host 首次写入补全逻辑 +# --------------------------------------------------------------------------- # + + +class TestSaveBestHost: + def test_first_write_completes_defaults(self, isolated_config: Path) -> None: + cfg.save_best_host("180.153.18.170") + data = json.loads((isolated_config / "config.json").read_text("utf-8")) + assert data["best_host"] == "180.153.18.170" + # 首次写入应补全所有默认字段 + assert data["known_hosts"] == list(cfg._FALLBACK_HOSTS) + assert data["calc_hosts"] == list(cfg._FALLBACK_CALC_HOSTS) + assert data["mac_hosts"] == list(cfg._FALLBACK_MAC_HOSTS) + assert data["ex_hosts"] == list(cfg._FALLBACK_EX_HOSTS) + assert data["mac_ex_hosts"] == list(cfg._FALLBACK_MAC_EX_HOSTS) + assert data["port"] == cfg._FALLBACK_PORT + assert "best_host_updated_at" in data + + def test_second_write_preserves_existing(self, isolated_config: Path) -> None: + # 预置已存在的 known_hosts，save_best_host 不应覆盖它 + existing = { + "known_hosts": ["custom.host"], + "port": 9999, + } + (isolated_config / "config.json").write_text(json.dumps(existing), "utf-8") + cfg.save_best_host("new.host") + data = json.loads((isolated_config / "config.json").read_text("utf-8")) + assert data["best_host"] == "new.host" + # 已有字段应保留，不被默认值覆盖 + assert data["known_hosts"] == ["custom.host"] + assert data["port"] == 9999 + # 但缺失的字段应补全 + assert "calc_hosts" in data + + def test_atomic_write(self, isolated_config: Path) -> None: + # 写入后不应残留 .tmp 文件（原子替换） + cfg.save_best_host("x.host") + assert not (isolated_config / "config.json.tmp").exists() + assert (isolated_config / "config.json").exists() diff --git a/tests/unit/test_ex_reconnect.py b/tests/unit/test_ex_reconnect.py new file mode 100644 index 0000000..46c356b --- /dev/null +++ b/tests/unit/test_ex_reconnect.py @@ -0,0 +1,193 @@ +"""扩展行情 client 的指数退避重连测试（审计 #2）。 + +之前 ex 家族（ExTdxClient/MacExClient/AsyncExTdxClient/AsyncMacExClient）的 _execute +只重连 1 次无退避，与 A 股/MAC 的 4 次退避不一致。本测试验证统一后的退避行为， +并确认 MacExClient 重连后会重新 _login()。 +""" + +from __future__ import annotations + +import asyncio +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from easy_tdx._reconnect import _RETRY_DELAYS +from easy_tdx.ex.client import AsyncExTdxClient, ExTdxClient +from easy_tdx.ex.commands.get_markets import GetExMarketsCmd +from easy_tdx.ex.mac_client import AsyncMacExClient, MacExClient +from easy_tdx.exceptions import TdxConnectionError + + +class TestExTdxClientReconnect: + def test_reconnect_succeeds_on_second_attempt(self) -> None: + """首次抛 TdxConnectionError，重连后第 1 次重试成功。""" + with patch("easy_tdx.ex.client.ExTdxConnection") as mock_conn_cls: + mock_conn = MagicMock() + mock_conn.execute.side_effect = [TdxConnectionError("down"), ["market"]] + mock_conn_cls.return_value = mock_conn + + client = ExTdxClient("1.1.1.1", auto_reconnect=True) + with patch("easy_tdx.ex.client.time.sleep"): + result = client._execute(GetExMarketsCmd()) + assert result == ["market"] + assert mock_conn.close.call_count == 1 # 重连了 1 次 + + def test_all_retries_exhausted_raises_last(self) -> None: + """4 次重试全部失败，应抛出异常，且 sleep 4 次。""" + with patch("easy_tdx.ex.client.ExTdxConnection") as mock_conn_cls: + mock_conn = MagicMock() + mock_conn.execute.side_effect = TdxConnectionError("always down") + mock_conn_cls.return_value = mock_conn + + client = ExTdxClient("1.1.1.1", auto_reconnect=True) + with patch("easy_tdx.ex.client.time.sleep") as mock_sleep: + with pytest.raises(TdxConnectionError): + client._execute(GetExMarketsCmd()) + assert mock_sleep.call_count == len(_RETRY_DELAYS) + + def test_no_reconnect_when_disabled(self) -> None: + with patch("easy_tdx.ex.client.ExTdxConnection") as mock_conn_cls: + mock_conn = MagicMock() + mock_conn.execute.side_effect = TdxConnectionError("down") + mock_conn_cls.return_value = mock_conn + + client = ExTdxClient("1.1.1.1", auto_reconnect=False) + with patch("easy_tdx.ex.client.time.sleep") as mock_sleep: + with pytest.raises(TdxConnectionError): + client._execute(GetExMarketsCmd()) + mock_sleep.assert_not_called() + + +class TestMacExClientReconnect: + def test_reconnect_relogs_in(self) -> None: + """MacExClient 每次重连后必须重新 _login()（MAC 协议特有）。""" + with patch("easy_tdx.ex.mac_client.ExTdxConnection") as mock_conn_cls: + mock_conn = MagicMock() + mock_conn.execute.side_effect = [TdxConnectionError("down"), ["market"]] + mock_conn_cls.return_value = mock_conn + + client = MacExClient("1.1.1.1", auto_reconnect=True) + with ( + patch("easy_tdx.ex.mac_client.time.sleep"), + patch.object(client, "_login") as mock_login, + ): + result = client._execute(GetExMarketsCmd()) + assert result == ["market"] + # 重连 1 次应触发 1 次 _login + assert mock_login.call_count == 1 + + def test_all_retries_relogin_each_time(self) -> None: + """4 次重试全失败时，每次重连都应 _login()（共 4 次）。""" + with patch("easy_tdx.ex.mac_client.ExTdxConnection") as mock_conn_cls: + mock_conn = MagicMock() + mock_conn.execute.side_effect = TdxConnectionError("always down") + mock_conn_cls.return_value = mock_conn + + client = MacExClient("1.1.1.1", auto_reconnect=True) + with ( + patch("easy_tdx.ex.mac_client.time.sleep"), + patch.object(client, "_login") as mock_login, + ): + with pytest.raises(TdxConnectionError): + client._execute(GetExMarketsCmd()) + assert mock_login.call_count == len(_RETRY_DELAYS) + + +class TestAsyncExTdxClientReconnect: + def test_async_all_retries_exhausted(self) -> None: + async def main() -> None: + with patch("easy_tdx.ex.client.AsyncExTdxConnection") as mock_conn_cls: + mock_conn = MagicMock() + + async def _execute(cmd: object) -> list[str]: + raise TdxConnectionError("always down") + + mock_conn.execute = _execute + + async def _noop() -> None: + return None + + mock_conn.close = _noop + mock_conn.connect = _noop + mock_conn_cls.return_value = mock_conn + + client = AsyncExTdxClient("1.1.1.1", auto_reconnect=True, heartbeat_interval=0) + with patch("easy_tdx.ex.client.asyncio.sleep") as mock_sleep: + with pytest.raises(TdxConnectionError): + await client._execute(GetExMarketsCmd()) + assert mock_sleep.call_count == len(_RETRY_DELAYS) + + asyncio.run(main()) + + +class TestAsyncMacExClientReconnect: + def test_async_relogin_each_retry(self) -> None: + """AsyncMacExClient 每次重连后必须重新 _login()（覆盖 async relogin 路径）。""" + + async def main() -> None: + with patch("easy_tdx.ex.mac_client.AsyncExTdxConnection") as mock_conn_cls: + mock_conn = MagicMock() + + async def _execute(cmd: object) -> list[str]: + raise TdxConnectionError("always down") + + mock_conn.execute = _execute + + async def _noop() -> None: + return None + + mock_conn.close = _noop + mock_conn.connect = _noop + mock_conn_cls.return_value = mock_conn + + client = AsyncMacExClient("1.1.1.1", auto_reconnect=True, heartbeat_interval=0) + with ( + patch("easy_tdx.ex.mac_client.asyncio.sleep"), + patch.object(client, "_login", new_callable=AsyncMock) as mock_login, + ): + with pytest.raises(TdxConnectionError): + await client._execute(GetExMarketsCmd()) + # 4 次重连应触发 4 次 _login + assert mock_login.call_count == len(_RETRY_DELAYS) + + asyncio.run(main()) + + +class TestBackoffDelayValues: + """验证退避延迟值序列与 _RETRY_DELAYS 完全一致（防硬编码回归）。""" + + def test_sync_ex_uses_exact_delays(self) -> None: + with patch("easy_tdx.ex.client.ExTdxConnection") as mock_conn_cls: + mock_conn = MagicMock() + mock_conn.execute.side_effect = TdxConnectionError("down") + mock_conn_cls.return_value = mock_conn + + client = ExTdxClient("1.1.1.1", auto_reconnect=True) + with patch("easy_tdx.ex.client.time.sleep") as mock_sleep: + with pytest.raises(TdxConnectionError): + client._execute(GetExMarketsCmd()) + actual = tuple(c.args[0] for c in mock_sleep.call_args_list) + assert actual == _RETRY_DELAYS + + +class TestMacExLoginRetriedOnConnectionError: + """登录握手期抛 TdxConnectionError 应继续重试（验证 _login 纳入 inner try）。""" + + def test_login_conn_error_triggers_full_retry(self) -> None: + """_login 抛 TdxConnectionError 时不应逃逸，应跑完 4 次重试。""" + with patch("easy_tdx.ex.mac_client.ExTdxConnection") as mock_conn_cls: + mock_conn = MagicMock() + mock_conn.execute.side_effect = TdxConnectionError("always down") + mock_conn_cls.return_value = mock_conn + + client = MacExClient("1.1.1.1", auto_reconnect=True) + # _login 抛 TdxConnectionError（模拟登录握手期连接又断） + with ( + patch("easy_tdx.ex.mac_client.time.sleep") as mock_sleep, + patch.object(client, "_login", side_effect=TdxConnectionError("login lost")), + ): + with pytest.raises(TdxConnectionError): + client._execute(GetExMarketsCmd()) + # 关键：_login 异常被纳入重试，4 次都跑了（而非第 1 次就逃逸） + assert mock_sleep.call_count == len(_RETRY_DELAYS) diff --git a/tests/unit/test_public_api.py b/tests/unit/test_public_api.py new file mode 100644 index 0000000..941bdcc --- /dev/null +++ b/tests/unit/test_public_api.py @@ -0,0 +1,137 @@ +"""公共 API 导出完整性测试 —— 防止 __all__ 与实际导出漂移（审计 #13）。 + +确保 easy_tdx.__all__ 中每个名字都能从顶层包成功导入， +且文档中描述的模型（FundFlow/MarketStat 等）确实可访问。 + +复审补充（L3）：进一步断言导出对象的**类型**，避免类名被意外绑成模块、 +None、或常量。仅"可导入"不足以守住类型契约。 +""" + +from __future__ import annotations + +import inspect + +import easy_tdx + +# 期望的导出契约：每个公共名字应对应的对象类型。 +# - "class" → 必须 inspect.isclass（client / 枚举 / 数据模型 / 异常） +# - "func" → 必须 callable 且非 class（ping_* / save_best_*） +# - "constant" → 兜底（KNOWN_HOSTS / XDXR_CATEGORY_NAMES 等映射表或常量） +_EXPECTED_KIND: dict[str, str] = { + # client 类 + "TdxClient": "class", + "AsyncTdxClient": "class", + "MacClient": "class", + "AsyncMacClient": "class", + "MacExClient": "class", + "AsyncMacExClient": "class", + "ExTdxClient": "class", + "AsyncExTdxClient": "class", + "UnifiedTdxClient": "class", + "AsyncUnifiedTdxClient": "class", + # 枚举 + "Market": "class", + "KlineCategory": "class", + "Adjust": "class", + "BoardType": "class", + "Category": "class", + "ExMarket": "class", + "FilterType": "class", + "Period": "class", + "SortOrder": "class", + "SortType": "class", + # 数据模型 + "SecurityBar": "class", + "SecurityQuote": "class", + "SecurityInfo": "class", + "MinuteBar": "class", + "TransactionRecord": "class", + "XdxrRecord": "class", + "FinanceInfo": "class", + "CompanyInfoCategory": "class", + "FinancialFileInfo": "class", + "FinancialRecord": "class", + "TdxBlock": "class", + "MarketStat": "class", + "FundFlow": "class", + "HistoricalFundFlow": "class", + # 异常 + "TdxError": "class", + "TdxConnectionError": "class", + "TdxDecodeError": "class", + "TdxCommandError": "class", + # 函数 + "ping_all": "func", + "ping_mac_all": "func", + "save_best_host": "func", + "save_best_ex_host": "func", + # 常量 / 映射表 + "KNOWN_EX_HOSTS": "constant", + "KNOWN_HOSTS": "constant", + "CALC_HOSTS": "constant", + "MAC_HOSTS": "constant", + "XDXR_CATEGORY_NAMES": "constant", +} + + +def test_all_names_are_importable() -> None: + """__all__ 里每个名字都必须能从 easy_tdx 顶层获取到非 None 对象。""" + missing = [name for name in easy_tdx.__all__ if getattr(easy_tdx, name, None) is None] + assert missing == [], f"__all__ 中以下名字无法从 easy_tdx 导入: {missing}" + + +def test_expected_kind_contract_is_complete() -> None: + """_EXPECTED_KIND 必须覆盖 __all__ 的每个名字，否则契约会悄悄漂移（审计复审 L3）。""" + covered = set(_EXPECTED_KIND) + exported = set(easy_tdx.__all__) + missing_kind = exported - covered + extra_kind = covered - exported + assert not missing_kind, f"以下导出未在 _EXPECTED_KIND 中声明类型契约: {sorted(missing_kind)}" + assert not extra_kind, f"_EXPECTED_KIND 含未导出的名字（已移除？）: {sorted(extra_kind)}" + + +def test_exported_objects_have_expected_type() -> None: + """断言每个导出对象的类型符合契约（审计复审 L3）。 + + 防止类名被绑成模块/None/常量——仅"可导入"不足以守住类型。 + """ + wrong: list[str] = [] + for name, kind in _EXPECTED_KIND.items(): + obj = getattr(easy_tdx, name, None) + if obj is None: + wrong.append(f"{name}: 不应为 None") + continue + if kind == "class": + if not inspect.isclass(obj): + wrong.append(f"{name}: 期望 class，实际 {type(obj).__name__}") + elif kind == "func": + # callable 但不能是 class（避免类被当成函数） + if not callable(obj) or inspect.isclass(obj): + wrong.append(f"{name}: 期望 function，实际 {type(obj).__name__}") + # "constant" 兜底，不做严格断言 + assert wrong == [], "导出对象类型契约违反: \n" + "\n".join(wrong) + + +def test_documented_models_exported() -> None: + """api_reference.md 文档描述的模型必须在公共导出中（审计 #13）。""" + for name in ("FundFlow", "MarketStat", "HistoricalFundFlow", "TdxBlock"): + assert name in easy_tdx.__all__, f"{name} 应在 easy_tdx.__all__ 中" + assert inspect.isclass(getattr(easy_tdx, name)), f"{name} 应是类" + + +def test_core_clients_exported() -> None: + """8 个 client 类与门面都应导出且确实是类（审计 #13 + 复审 L3）。""" + for name in ( + "TdxClient", + "AsyncTdxClient", + "MacClient", + "AsyncMacClient", + "ExTdxClient", + "AsyncExTdxClient", + "MacExClient", + "AsyncMacExClient", + "UnifiedTdxClient", + "AsyncUnifiedTdxClient", + ): + assert name in easy_tdx.__all__, f"{name} 应在 easy_tdx.__all__ 中" + assert inspect.isclass(getattr(easy_tdx, name)), f"{name} 应是类" diff --git a/tests/unit/test_scanner.py b/tests/unit/test_scanner.py index 4faa33c..8a2db15 100644 --- a/tests/unit/test_scanner.py +++ b/tests/unit/test_scanner.py @@ -210,3 +210,57 @@ class TestIncrementalScan: codes2 = sorted(r.code for r in results2) # 结果可能相同 (策略没变), 但不应崩溃 assert len(codes2) >= 1 + + +class TestScanFailureLogging: + """扫描失败日志回归（审计复审 L2）。 + + 首轮 #6 将扫描循环的 ``except Exception: continue`` 评为"系统性失败被静默 + 吞掉"。复审 L2 修复：单股失败记录 warning + 失败计数，失败率超阈值时 + 循环结束发出 summary。这些测试用 monkeypatch 让 ``_scan_one`` 抛错模拟 + 损坏 .day / 策略异常等场景，断言失败被记录（read_daily_bars 本身对短文件 + 容错返回 0 条，不会抛错，故用 monkeypatch 构造确定性失败）。 + """ + + def test_serial_scan_logs_per_stock_failure( + self, vipdoc: Path, caplog: pytest.LogCaptureFixture, monkeypatch: pytest.MonkeyPatch + ) -> None: + """单股 _scan_one 抛错时，串行扫描应记录 warning（审计复审 L2）。""" + scanner = SignalScanner(AlwaysBuyStrategy, vipdoc_path=vipdoc) + + def _boom(self: SignalScanner, filepath: Path, market: str, code: str) -> None: + raise RuntimeError(f"simulated corrupt day for {code}") + + monkeypatch.setattr(SignalScanner, "_scan_one", _boom) + + with caplog.at_level("WARNING", logger="easy_tdx.screen.scanner"): + results = scanner.scan(universe="all", workers=0) + + # 全部抛错 → 无结果，但不崩溃（容错语义：跳过继续） + assert results == [] + # 每个被扫描的 A 股都应有一条 warning + warnings = [r for r in caplog.records if r.levelname == "WARNING"] + assert len(warnings) >= 1, "单股失败应触发 warning 日志" + assert any("失败" in r.getMessage() for r in warnings) + + def test_serial_scan_high_failure_rate_emits_summary( + self, vipdoc: Path, caplog: pytest.LogCaptureFixture, monkeypatch: pytest.MonkeyPatch + ) -> None: + """失败率超阈值时应发出汇总告警（审计复审 L2）。 + + 全部 A 股 _scan_one 抛错（失败率 100% > 50% 阈值），断言扫描完成后 + 有一条 summary warning。 + """ + scanner = SignalScanner(AlwaysBuyStrategy, vipdoc_path=vipdoc) + + def _boom(self: SignalScanner, filepath: Path, market: str, code: str) -> None: + raise RuntimeError(f"simulated corrupt day for {code}") + + monkeypatch.setattr(SignalScanner, "_scan_one", _boom) + + with caplog.at_level("WARNING", logger="easy_tdx.screen.scanner"): + scanner.scan(universe="all", workers=0) + + # 应有汇总告警提到"失败率过高" + summary_msgs = [r.getMessage() for r in caplog.records if "失败率" in r.getMessage()] + assert summary_msgs, "失败率过高时应发出汇总 warning"