From 15cc7680c4bb34214daf5a9227e23611c87691cd Mon Sep 17 00:00:00 2001 From: Justin Gu <97915@qq.com> Date: Thu, 11 Jun 2026 03:57:48 +0800 Subject: [PATCH] =?UTF-8?q?release:=20v1.9.7=20=E2=80=94=20CLI=E5=85=A8?= =?UTF-8?q?=E9=87=8F=E9=9B=86=E6=88=90=EF=BC=88workers/cache/chanlun-level?= =?UTF-8?q?/portfolio/multi-level=EF=BC=89+=20bugfix?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 66 +++++++++++++ pyproject.toml | 2 +- src/easy_tdx/backtest/cli.py | 161 ++++++++++++++++++++++++++++++++ src/easy_tdx/cli/__init__.py | 3 +- src/easy_tdx/cli/cmd_chanlun.py | 113 +++++++++++++++++++++- src/easy_tdx/screen/cli.py | 24 ++++- 6 files changed, 361 insertions(+), 8 deletions(-) diff --git a/README.md b/README.md index 6d0962a..277f747 100644 --- a/README.md +++ b/README.md @@ -160,6 +160,10 @@ easy-tdx indicator RSI -m SZ -c 000001 --no-ohlcv easy-tdx chanlun SZ 000001 --table easy-tdx chanlun SH 600519 --adjust QFQ --table easy-tdx chanlun SZ 000001 --period 30MIN + +# 多级别联立：分析日线最后一笔在 30 分钟级别中的走势结构 +easy-tdx chanlun SZ 000001 --multi-level 30MIN --table +easy-tdx chanlun SH 600519 --multi-level 5MIN ``` #### 输出示例 @@ -276,6 +280,9 @@ easy-tdx chanlun SZ 000001 --period 30MIN ```bash easy-tdx backtest SZ 300308 --strategy-file strategies/expma_cross.py --count 2000 --cash 1000000 --adjust QFQ --table # 推荐加上 --slippage 0.01 模拟真实滑点（元/股），使回测更贴近实盘 + +# 缠论自动桥接：引擎自动计算缠论分析并注入策略 self.chanlun +easy-tdx backtest SZ 000001 --strategy-file strategies/chanlun_strategy.py --chanlun-level DAILY --table ``` 输出示例： @@ -364,6 +371,45 @@ for r in results[:5]: `--show` 会用 matplotlib 弹出一个双轴对比窗口：左轴蓝色线是归一化股价，右轴红色线是最佳策略的资金曲线，绿三角=买入、黄三角=卖出，标题显示股票名称和关键绩效指标。需要 `pip install matplotlib`。 +**多标的组合回测（portfolio）：** + +`easy-tdx portfolio` 对多只股票同时回测，共享资金池，按均等比例分配，汇总组合整体绩效： + +```bash +# 两只股票组合回测 +easy-tdx portfolio --stocks SZ:000001,SH:600519 --strategy-file strategies/ma_cross.py --table + +# 自定义资金和周期 +easy-tdx portfolio --stocks SZ:000001,SH:600519,SH:600036 \ + --strategy-file strategies/expma_cross.py --cash 500000 --period DAILY --count 1000 --table + +# 搭配缠论桥接 +easy-tdx portfolio --stocks SZ:000001,SH:600519 \ + --strategy-file strategies/chanlun_strategy.py --chanlun-level DAILY --table +``` + +输出示例： + +``` +=== 组合回测绩效概要 === +标的数量: 3 +总资金: 200,000 +组合收益率: 28.50% +组合年化: 28.50% + +── 各标的详情 ── + SZ000001: 收益=35.20% 夏普=0.92 回撤=15.30% 分配=33% 交易=12 + SH600519: 收益=18.40% 夏普=0.68 回撤=8.50% 分配=33% 交易=8 + SH600036: 收益=31.90% 夏普=0.85 回撤=12.10% 分配=33% 交易=15 +``` + +| 参数 | 说明 | +|------|------| +| `--stocks` | 股票列表：逗号分隔的 `市场:代码`（如 `SZ:000001,SH:600519`） | +| `--cash` | 总资金（默认 20 万） | +| `--allocation` | 资金分配方式（目前支持 `equal` 均等分配） | +| `--chanlun-level` | 自动计算缠论分析并注入策略（如 DAILY/30MIN） | + 输出示例（以 SZ 300308 为例）： ``` @@ -496,6 +542,12 @@ easy-tdx screen scan --strategy strategies/macd_cross.py --universe sz --output # 从自定义股票列表扫描 easy-tdx screen scan --strategy strategies/bollinger_breakout.py --universe my_stocks.txt --output signals.json + +# 并发扫描（推荐 4-8 进程，速度提升 4-8 倍） +easy-tdx screen scan --strategy strategies/rsi_reversal.py --workers 4 --output signals.json + +# 增量扫描（缓存未修改的 .day 文件，跳过重复计算） +easy-tdx screen scan --strategy strategies/rsi_reversal.py --cache scan_cache.json --output signals.json ``` 输出示例（JSON）： @@ -543,6 +595,8 @@ easy-tdx screen rank --from signals.json --sort sharpe --top 10 --table --names |------|------| | `--universe` | `all`（默认，沪深全 A）/ `sh` / `sz` / 文件路径（每行 "市场 代码"） | | `--vipdoc` | 离线数据目录（默认自动检测通达信安装路径） | +| `--workers` | 并发进程数：`0` 串行（默认）/ `2+` ProcessPoolExecutor 并发（推荐 4-8） | +| `--cache` | 增量扫描缓存文件路径（JSON，mtime 未变的文件自动跳过） | | `--sort` | 排序指标：`sharpe`（默认）/ `total_return` / `max_drawdown` / `win_rate` 等 | | `--sort-reverse` | 升序（用于回撤等越小越好的指标） | | `--names` | 在线补齐股票名称（默认关闭，只查排名中的几十只） | @@ -700,6 +754,7 @@ easy-tdx offline sync-all | `indicator` | 技术指标计算（32 个：MACD/KDJ/RSI/BOLL/DMI/ATR...） | | `indicator-list` | 列出可用技术指标 | | `backtest` | 回测引擎（加载策略文件，输出绩效报告） | +| `portfolio` | 多标的组合回测（共享资金池，均等分配，汇总绩效） | | `run-all` | 批量运行所有策略并排名（绩效排名 + 综合评分 + 可选图表） | | `screen scan` | 策略选股扫描（纯离线，全市场信号扫描） | | `screen rank` | 扫描结果回测排名（按夏普/回撤等指标排序） | @@ -1253,6 +1308,17 @@ ruff format --check src/ tests/ # format check ## Changelog +### 1.9.7 (2026-06-11) + +**CLI 全量集成** — v1.9.6 新增的 6 项功能全部暴露到 CLI，修复缠论多级别联立的 client 生命周期 bug。 + +- **`screen scan` 并发扫描**：新增 `--workers N` 参数，ProcessPoolExecutor 并行处理，推荐 4-8 进程，扫描速度提升 4-8 倍 +- **`screen scan` 增量缓存**：新增 `--cache PATH` 参数，mtime 检测未修改的 `.day` 文件自动跳过 +- **`backtest` 缠论桥接**：新增 `--chanlun-level LEVEL` 参数，引擎自动计算缠论分析并注入策略 `self.chanlun` +- **`portfolio` 组合回测**：新增 `easy-tdx portfolio` 命令，多标的共享资金池、均等分配、汇总绩效 +- **`chanlun` 多级别联立**：新增 `--multi-level PERIOD` 参数，分析高级别最后一笔在低级别中的趋势方向、笔重叠、背驰条件 +- **Bug 修复**：`cmd_chanlun.py` 中 `_run_multi_level` 在 `with` 块外使用 `client`，导致已关闭连接报错 + ### 1.9.6 (2026-06-11) **工程质量全面升级** — 基于 Devin AI 代码审查的 12 项改进建议全部落地，覆盖 CI、回测引擎、缠论模块、扫描引擎和架构层面。 diff --git a/pyproject.toml b/pyproject.toml index 42d7eb9..175c288 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "easy-tdx" -version = "1.9.6" +version = "1.9.7" description = "通达信 TCP 协议行情数据客户端，支持在线行情、离线数据读取与写入同步" readme = "README.md" requires-python = ">=3.10" diff --git a/src/easy_tdx/backtest/cli.py b/src/easy_tdx/backtest/cli.py index 56134c8..6a31821 100644 --- a/src/easy_tdx/backtest/cli.py +++ b/src/easy_tdx/backtest/cli.py @@ -39,6 +39,12 @@ import click @click.option("--adjust", default="NONE", help="复权: NONE/QFQ/HFQ") @click.option("--count", default=500, type=int, help="K线数量") @click.option("--indicators", default=None, help="预计算指标（逗号分隔）") +@click.option( + "--chanlun-level", + "chanlun_level", + default=None, + help="自动计算缠论分析并注入策略（如 DAILY/30MIN）", +) @click.option("--table", "use_table", is_flag=True, help="表格输出") @click.option("--output", "output_fmt", type=click.Choice(["json", "table", "csv"]), default="json") def backtest( @@ -55,6 +61,7 @@ def backtest( adjust: str, count: int, indicators: str | None, + chanlun_level: str | None, use_table: bool, output_fmt: str, ) -> None: @@ -68,6 +75,8 @@ def backtest( easy-tdx backtest SZ 000001 --strategy-file my_strategy.py --indicators MACD,KDJ + easy-tdx backtest SZ 000001 --strategy-file chanlun_strategy.py --chanlun-level DAILY + easy-tdx backtest SZ 000001 \ --combo-strategies strategies/macd_cross.py,strategies/rsi_reversal.py \ --combo-mode MAJORITY --table @@ -128,6 +137,7 @@ def backtest( cash=cash, commission=commission, execution=execution, + chanlun_level=chanlun_level, ) result = engine.run(df) @@ -250,6 +260,8 @@ def _print_table(result: Any) -> None: click.echo(f"初始资金: {config.get('cash', 0):.2f}") click.echo(f"佣金率: {config.get('commission', 0):.4f}") click.echo(f"成交规则: {config.get('execution', 'next_open')}") + if config.get("chanlun_level"): + click.echo(f"缠论级别: {config.get('chanlun_level')}") click.echo() if config.get("future_leak_warning"): @@ -269,3 +281,152 @@ def _print_table(result: Any) -> None: ) else: click.echo("无交易记录") + + +# ── portfolio 多标的组合回测命令 ───────────────────────────────────────────── + + +@click.command() +@click.option( + "--stocks", + required=True, + help="股票列表：逗号分隔的 市场:代码（如 SZ:000001,SH:600519,SH:600036）", +) +@click.option("--strategy-file", "strategy_file", required=True, help="Python 策略文件路径") +@click.option("--cash", default=200_000.0, type=float, help="总资金（默认 20 万）") +@click.option("--commission", default=0.0003, type=float, help="佣金率") +@click.option( + "--execution", + default="next_open", + type=click.Choice(["next_open", "next_close", "this_close", "worst", "best"]), + help="成交价规则", +) +@click.option("--period", default="DAILY", help="K线周期") +@click.option("--adjust", default="NONE", help="复权: NONE/QFQ/HFQ") +@click.option("--count", default=500, type=int, help="K线数量") +@click.option( + "--allocation", + default="equal", + type=click.Choice(["equal"], case_sensitive=False), + help="资金分配方式（默认 equal 均等分配）", +) +@click.option( + "--chanlun-level", + "chanlun_level", + default=None, + help="自动计算缠论分析并注入策略（如 DAILY/30MIN）", +) +@click.option("--table", "use_table", is_flag=True, help="表格输出") +@click.option("--output", "output_fmt", type=click.Choice(["json", "table", "csv"]), default="json") +def portfolio( + stocks: str, + strategy_file: str, + cash: float, + commission: float, + execution: str, + period: str, + adjust: str, + count: int, + allocation: str, + chanlun_level: str | None, + use_table: bool, + output_fmt: str, +) -> None: + """多标的组合回测：共享资金池，独立产生信号，统一管理仓位。 + + 对多只股票同时回测，按均等比例分配资金，汇总组合整体绩效。 + + 示例： + + easy-tdx portfolio --stocks SZ:000001,SH:600519 --strategy-file ma_cross.py + + easy-tdx portfolio --stocks SZ:000001,SH:600519,SH:600036 \\ + --strategy-file my_strategy.py --cash 500000 --table + + easy-tdx portfolio --stocks SZ:000001,SH:600519 \\ + --strategy-file chanlun_strat.py --chanlun-level DAILY + """ + import json + + from ..cli.conn import get_mac_client + from ..cli.parsers import parse_adjust, parse_market, parse_period + from .portfolio_engine import PortfolioBacktestEngine, StockData + + # 1. 加载策略 + strategy_cls = _load_strategy_from_file(strategy_file) + strategy_name = strategy_cls.__name__ + + # 2. 解析股票列表 + stock_list = [] + for item in stocks.split(","): + item = item.strip() + if ":" not in item: + click.echo(f"错误: 股票格式应为 市场:代码，如 SZ:000001，收到: {item}", err=True) + raise SystemExit(1) + mkt_str, code = item.split(":", 1) + stock_list.append((mkt_str.strip().upper(), code.strip())) + + if not stock_list: + click.echo("错误: 未指定股票", err=True) + raise SystemExit(1) + + click.echo(f"策略: {strategy_name} | 标的: {len(stock_list)} 只 | 资金: {cash:,.0f}", err=True) + + # 3. 获取数据 + stock_data_list: list[StockData] = [] + with get_mac_client() as client: + for mkt_str, code in stock_list: + mkt = parse_market(mkt_str) + df = client.get_stock_kline( + mkt, + code, + period=parse_period(period), + start=0, + count=count, + adjust=parse_adjust(adjust), + ) + stock_data_list.append(StockData(code=code, market=mkt_str, df=df)) + + # 4. 创建引擎并运行 + engine = PortfolioBacktestEngine( + strategy_cls=strategy_cls, + stocks=stock_data_list, + total_cash=cash, + allocation=allocation, + commission=commission, + execution=execution, + chanlun_level=chanlun_level, + ) + result = engine.run() + + # 5. 输出结果 + fmt = "table" if use_table else output_fmt + if fmt == "table": + _print_portfolio_table(result) + else: + click.echo(json.dumps(result.to_dict(), ensure_ascii=False, indent=2)) + + +def _print_portfolio_table(result: Any) -> None: + """以表格形式输出组合回测结果。""" + perf = result.total_performance + + click.echo("=== 组合回测绩效概要 ===") + click.echo(f"标的数量: {perf.get('total_stocks', 0)}") + click.echo(f"总资金: {perf.get('total_cash', 0):,.0f}") + click.echo(f"组合收益率: {perf.get('total_return', 0):.2%}") + click.echo(f"组合年化: {perf.get('annual_return', 0):.2%}") + click.echo() + + click.echo("── 各标的详情 ──") + for key, stock_result in result.individual_results.items(): + sp = stock_result.performance + alloc = result.equity_allocation.get(key, 0) + click.echo( + f" {key}: 收益={sp.get('total_return', 0):.2%} " + f"夏普={sp.get('sharpe', 0):.2f} " + f"回撤={sp.get('max_drawdown', 0):.2%} " + f"分配={alloc:.0%} " + f"交易={sp.get('total_trades', 0)}" + ) + click.echo() diff --git a/src/easy_tdx/cli/__init__.py b/src/easy_tdx/cli/__init__.py index 302e49f..87b3de0 100644 --- a/src/easy_tdx/cli/__init__.py +++ b/src/easy_tdx/cli/__init__.py @@ -4,7 +4,7 @@ from __future__ import annotations import click -from ..backtest.cli import backtest +from ..backtest.cli import backtest, portfolio from ..screen.cli import screen from .cmd_admin import ping, version from .cmd_auction import auction @@ -73,5 +73,6 @@ cli.add_command(indicator_list) cli.add_command(offline) cli.add_command(chanlun) cli.add_command(backtest) +cli.add_command(portfolio) cli.add_command(run_all) cli.add_command(screen) diff --git a/src/easy_tdx/cli/cmd_chanlun.py b/src/easy_tdx/cli/cmd_chanlun.py index 43f4ad1..762889b 100644 --- a/src/easy_tdx/cli/cmd_chanlun.py +++ b/src/easy_tdx/cli/cmd_chanlun.py @@ -16,6 +16,12 @@ import click ) @click.option("--count", default=800, type=int, help="K线数量") @click.option("--adjust", default="NONE", help="复权: NONE/QFQ/HFQ") +@click.option( + "--multi-level", + "low_level_period", + default=None, + help="低级别周期（多级别联立），如 30MIN；分析高级别最后一笔在低级别中的走势", +) @click.option("--table", "use_table", is_flag=True, help="表格输出") @click.option("--output", "output_fmt", type=click.Choice(["json", "table", "csv"]), default="json") def chanlun( @@ -24,6 +30,7 @@ def chanlun( period: str, count: int, adjust: str, + low_level_period: str | None, use_table: bool, output_fmt: str, ) -> None: @@ -36,6 +43,10 @@ def chanlun( easy-tdx chanlun SH 600519 --adjust QFQ --table easy-tdx chanlun SZ 000001 --period 30MIN + + easy-tdx chanlun SZ 000001 --multi-level 30MIN --table + + easy-tdx chanlun SZ 000001 --multi-level 5MIN """ from ..chanlun.analyser import ChanlunAnalyser from .conn import get_mac_client @@ -52,10 +63,19 @@ def chanlun( adjust=parse_adjust(adjust), ) - analyser = ChanlunAnalyser(code=code, frequency=period) - result = analyser.process_klines(df) + analyser = ChanlunAnalyser(code=code, frequency=period) + result = analyser.process_klines(df) - result_dict = result.to_dict() + result_dict = result.to_dict() + + # 多级别联立分析（需要在 with 块内使用 client 获取低级别数据） + multi_level_info: dict[str, Any] | None = None + if low_level_period is not None: + multi_level_info = _run_multi_level( + client, mkt, code, period, low_level_period, count, adjust, analyser, result + ) + if multi_level_info is not None: + result_dict["multi_level"] = multi_level_info fmt = "table" if use_table else output_fmt if fmt == "json": @@ -126,3 +146,90 @@ def _print_table(result: dict[str, Any]) -> None: for bc in result["bcs"]: status = "✓" if bc["bc"] else "✗" click.echo(f" [{status}] {bc['type']}: {bc['msg']}") + + if result.get("multi_level"): + ml = result["multi_level"] + click.echo("── 多级别联立 ──") + click.echo(f" 高级别: {ml.get('high_level', '?')} 低级别: {ml.get('low_level', '?')}") + qs = ml.get("low_level_qs", {}) + if qs: + direction = qs.get("trend_direction") + dir_str = {"up": "↑ 上升", "down": "↓ 下降"}.get(str(direction), "— 盘整") + click.echo(f" 低级别笔: {qs.get('bi_count', 0)} 中枢: {qs.get('zs_count', 0)}") + click.echo( + f" 趋势方向: {dir_str} " + f"趋势: {'是' if qs.get('has_trend') else '否'} " + f"盘整: {'是' if qs.get('has_consolidation') else '否'}" + ) + click.echo( + f" 笔重叠: {'是' if qs.get('bi_overlap') else '否'} " + f"背驰可能: {'是' if qs.get('divergence_possible') else '否'}" + ) + click.echo() + + +def _run_multi_level( + client: Any, + mkt: Any, + code: str, + high_period: str, + low_period: str, + count: int, + adjust: str, + high_analyser: Any, + high_result: Any, +) -> dict[str, Any] | None: + """运行多级别联立分析。 + + 获取低级别数据，分析高级别最后一笔在低级别中的走势结构。 + + Args: + client: TdxClient 实例 + mkt: Market 枚举值 + code: 股票代码 + high_period: 高级别周期 + low_period: 低级别周期 + count: K线数量 + adjust: 复权方式 + high_analyser: 高级别分析器（已处理） + high_result: 高级别分析结果 + + Returns: + 多级别分析信息字典，或 None（数据不足时） + """ + from ..chanlun.analyser import ChanlunAnalyser + from ..chanlun.multi_level import MultiLevelAnalyser + from .parsers import parse_adjust, parse_period + + # 高级别最后一笔 + if not high_result.bis: + return None + + last_bi = high_result.bis[-1] + + # 获取低级别数据（需要更多 K 线来覆盖高级别笔的时间范围） + df_low = client.get_stock_kline( + mkt, + code, + period=parse_period(low_period), + start=0, + count=min(count * 8, 8000), # 低级别需要更多数据 + adjust=parse_adjust(adjust), + ) + + low_analyser = ChanlunAnalyser(code=code, frequency=low_period) + + mla = MultiLevelAnalyser() + mla.add_level("high", high_analyser) + mla.add_level("low", low_analyser) + # 高级别已经处理过，只需注册；低级别需要处理 + mla.process("low", df_low) + + qs = mla.query_low_level_qs("high", "low", last_bi) + + return { + "high_level": high_period, + "low_level": low_period, + "last_bi_index": last_bi.index if hasattr(last_bi, "index") else None, + "low_level_qs": qs, + } diff --git a/src/easy_tdx/screen/cli.py b/src/easy_tdx/screen/cli.py index 759b66d..0dac68a 100644 --- a/src/easy_tdx/screen/cli.py +++ b/src/easy_tdx/screen/cli.py @@ -38,6 +38,13 @@ def screen() -> None: @click.option("--vipdoc", default=None, help="离线数据目录（默认自动检测）") @click.option("--cash", default=100_000.0, type=float, help="初始资金") @click.option("--commission", default=0.0003, type=float, help="佣金率") +@click.option( + "--workers", + default=0, + type=int, + help="并发工作进程数: 0=串行（默认），2+=ProcessPoolExecutor 并发（推荐 4-8）", +) +@click.option("--cache", "cache_file", default=None, help="增量扫描缓存文件路径（JSON）") def scan( strategy_file: str, output_file: str | None, @@ -45,10 +52,12 @@ def scan( vipdoc: str | None, cash: float, commission: float, + workers: int, + cache_file: str | None, ) -> None: """纯离线扫描全市场，找出触发买入信号的股票。 - 读取本地通达信 .day 文件，零网络 IO，全市场约 30-60 秒。 + 读取本地通达信 .day 文件，零网络 IO，串行约 30-60 秒，并发可提速 4-8 倍。 示例： @@ -56,13 +65,21 @@ def scan( easy-tdx screen scan --strategy strategies/rsi_reversal.py --output signals.json - easy-tdx screen scan --strategy strategies/rsi_reversal.py --universe sz + easy-tdx screen scan --strategy strategies/rsi_reversal.py --universe sz --workers 4 + + easy-tdx screen scan --strategy strategies/rsi_reversal.py --cache scan_cache.json """ strategy_cls = _load_strategy(strategy_file) strategy_name = strategy_cls.__name__ click.echo(f"策略: {strategy_name}", err=True) click.echo(f"范围: {universe}", err=True) + if workers > 0: + click.echo(f"并发: {workers} 进程", err=True) + if workers > 0 and cache_file: + click.echo("注意: 并发模式暂不支持增量缓存，--cache 参数将被忽略", err=True) + if cache_file: + click.echo(f"缓存: {cache_file}", err=True) from .scanner import SignalScanner @@ -71,6 +88,7 @@ def scan( vipdoc_path=vipdoc, cash=cash, commission=commission, + cache_file=cache_file, ) # 进度回调（输出到 stderr，避免污染 stdout 的 JSON） @@ -85,7 +103,7 @@ def scan( pct = current * 100 // total if total > 0 else 0 click.echo(f"\r[{current}/{total}] {pct}% scanning {name}", nl=False, err=True) - results = scanner.scan(universe=universe, progress_callback=on_progress) + results = scanner.scan(universe=universe, progress_callback=on_progress, workers=workers) # 生成 JSON json_str = scanner.to_json(