diff --git a/README.md b/README.md index c2476ff..c475722 100644 --- a/README.md +++ b/README.md @@ -458,6 +458,75 @@ class MyStrategy(Strategy): 完整 API 参考：[docs/backtest_usage.md](docs/backtest_usage.md) +### 策略选股扫描（screen） + +把策略翻转成选股器：给定一个策略，扫描全市场找出今天触发买入信号的股票，再对这些信号做历史回测排名。**纯离线数据**，读取本地通达信 `.day` 文件，全市场约 30-60 秒。 + +两步走工作流： + +**第一步：信号扫描（scan）** + +```bash +# 扫描沪深全 A，找出 RSI 超卖触发的股票 +easy-tdx screen scan --strategy strategies/rsi_reversal.py --output signals.json + +# 缩小范围 +easy-tdx screen scan --strategy strategies/macd_cross.py --universe sz --output signals.json + +# 从自定义股票列表扫描 +easy-tdx screen scan --strategy strategies/bollinger_breakout.py --universe my_stocks.txt --output signals.json +``` + +输出示例（JSON）： + +```json +{ + "scan_time": "2026-06-10T18:30:00", + "strategy": "RSIStrategy", + "total_scanned": 4832, + "total_signals": 37, + "signals": [ + {"code": "000001", "market": "SZ", "signal_date": 20260610, "last_close": 12.35}, + {"code": "600519", "market": "SH", "signal_date": 20260610, "last_close": 1800.0} + ] +} +``` + +**第二步：回测排名（rank）** + +```bash +# 按夏普比率排名（默认） +easy-tdx screen rank --from signals.json --sort sharpe --top 20 --table + +# 按最大回撤排名（越小越好，用 --sort-reverse） +easy-tdx screen rank --from signals.json --sort max_drawdown --sort-reverse --table + +# 管道模式：一步到位 +easy-tdx screen scan --strategy strategies/rsi_reversal.py | easy-tdx screen rank --from - --table + +# 补齐股票名称（需要网络） +easy-tdx screen rank --from signals.json --sort sharpe --top 10 --table --names +``` + +输出示例（`--table`）： + +``` +[*] 信号排名 (按 sharpe 降序, 共 37 只) +══════════════════════════════════════════════════════════════════════════ +排名 代码 名称 总收益率 年化收益 最大回撤 夏普 胜率 交易 + *1 SZ300308 中际旭创 45.23% 18.72% 12.35% 1.85 62.5% 16 + *2 SH600519 贵州茅台 38.10% 15.90% 8.21% 1.62 58.3% 12 +``` + +| 参数 | 说明 | +|------|------| +| `--universe` | `all`（默认，沪深全 A）/ `sh` / `sz` / 文件路径（每行 "市场 代码"） | +| `--vipdoc` | 离线数据目录（默认自动检测通达信安装路径） | +| `--sort` | 排序指标：`sharpe`（默认）/ `total_return` / `max_drawdown` / `win_rate` 等 | +| `--sort-reverse` | 升序（用于回撤等越小越好的指标） | +| `--names` | 在线补齐股票名称（默认关闭，只查排名中的几十只） | +| `--count` | rank 使用最近 N 条 K 线（0=全部，默认 0） | + ### 捉妖大师（重点） 捉妖大师是多周期涨幅共振指标，通过 20/60/120 日涨幅及指数平滑判断短中长线趋势是否同向，用于筛选趋势刚启动的强势股。 @@ -610,6 +679,8 @@ easy-tdx offline sync-all | `indicator` | 技术指标计算（32 个：MACD/KDJ/RSI/BOLL/DMI/ATR...） | | `indicator-list` | 列出可用技术指标 | | `backtest` | 回测引擎（加载策略文件，输出绩效报告） | +| `screen scan` | 策略选股扫描（纯离线，全市场信号扫描） | +| `screen rank` | 扫描结果回测排名（按夏普/回撤等指标排序） | | `f10` | F10 公司信息 | | `fund-flow` | 历史资金流向 | | `ex kline` | 扩展市场 K 线 | @@ -1130,6 +1201,7 @@ src/easy_tdx/ ├── codec/ # price / volume / datetime / frame / bitmap 编解码 ├── chanlun/ # 缠论技术分析（K线合并/分型/笔/线段/中枢/买卖点/背驰） ├── backtest/ # 回测引擎（Strategy基类/向量化引擎/多因子组合/绩效分析） +├── screen/ # 策略选股扫描（scan信号扫描/rank回测排名） ├── models/ # 纯 dataclass，无业务逻辑 ├── offline/ # 离线数据读写模块（读取 + 写入同步） └── cli/ # easy-tdx CLI（click） @@ -1158,6 +1230,18 @@ ruff format --check src/ tests/ # format check ## Changelog +### 1.9.2 (2026-06-10) + +**策略选股扫描器** — 新增 `screen` 命令组，用策略扫描全市场找出触发买入信号的股票，再做历史回测排名。纯离线数据，零网络 IO。 + +- 新增 `screen scan` CLI 命令：纯离线扫描本地 `.day` 文件，提取策略信号，输出 JSON +- 新增 `screen rank` CLI 命令：读取扫描结果，批量回测并按夏普/回撤等指标排名 +- 新增 `src/easy_tdx/screen/` 模块：`SignalScanner`（扫描引擎）、`SignalRanker`（排名引擎） +- 两步走工作流：scan 几秒扫完全市场 → rank 对信号股做历史评估 +- 支持 `--universe` 指定范围（all/sh/sz/自定义文件）、`--sort` 排序、`--names` 在线补名称 +- 支持管道模式：`easy-tdx screen scan ... | easy-tdx screen rank --from - --table` +- 新增 20 个单元测试（离线，无需网络） + ### 1.9.0 (2026-06-10) **多因子组合回测** — 新增组合回测引擎，支持 2-3 个因子信号叠加，自动遍历所有组合寻找最优搭配。 diff --git a/pyproject.toml b/pyproject.toml index 7dcdfeb..1bd8400 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "easy-tdx" -version = "1.9.1" +version = "1.9.2" description = "通达信 TCP 协议行情数据客户端，支持在线行情、离线数据读取与写入同步" readme = "README.md" requires-python = ">=3.10" diff --git a/src/easy_tdx/cli/__init__.py b/src/easy_tdx/cli/__init__.py index 1095edb..a8d1db5 100644 --- a/src/easy_tdx/cli/__init__.py +++ b/src/easy_tdx/cli/__init__.py @@ -4,6 +4,8 @@ from __future__ import annotations import click +from ..backtest.cli import backtest +from ..screen.cli import screen from .cmd_admin import ping, version from .cmd_auction import auction from .cmd_board import belong_board, board_list, board_members, board_ranking, board_summary @@ -19,11 +21,13 @@ from .cmd_offline import offline from .cmd_quote import quote, quote_list from .cmd_tick import tick from .cmd_transaction import transaction -from ..backtest.cli import backtest @click.group() -@click.version_option(version=__import__("importlib.metadata", fromlist=["version"]).version("easy-tdx"), prog_name="easy-tdx") +@click.version_option( + version=__import__("importlib.metadata", fromlist=["version"]).version("easy-tdx"), + prog_name="easy-tdx", +) def cli() -> None: """easy-tdx -- 通达信行情数据 CLI（默认 JSON 输出，适合 Agent 使用）。 @@ -68,3 +72,4 @@ cli.add_command(indicator_list) cli.add_command(offline) cli.add_command(chanlun) cli.add_command(backtest) +cli.add_command(screen) diff --git a/src/easy_tdx/screen/__init__.py b/src/easy_tdx/screen/__init__.py new file mode 100644 index 0000000..27bd3d1 --- /dev/null +++ b/src/easy_tdx/screen/__init__.py @@ -0,0 +1,21 @@ +"""easy_tdx.screen — 策略选股扫描器。 + +两步走工作流： +1. scan: 用策略扫描全市场，找出触发买入信号的股票（纯离线） +2. rank: 对扫描结果做历史回测排名 + +用法:: + + # Step 1: 信号扫描 + easy-tdx screen scan --strategy strategies/rsi_reversal.py --output signals.json + + # Step 2: 回测排名 + easy-tdx screen rank --from signals.json --sort sharpe --top 20 --table +""" + +from easy_tdx.screen.scanner import ScanResult, SignalScanner # noqa: F401 + +__all__ = [ + "SignalScanner", + "ScanResult", +] diff --git a/src/easy_tdx/screen/cli.py b/src/easy_tdx/screen/cli.py new file mode 100644 index 0000000..759b66d --- /dev/null +++ b/src/easy_tdx/screen/cli.py @@ -0,0 +1,243 @@ +"""screen 命令组 — 策略选股扫描器 CLI。 + +子命令： + scan — 纯离线扫描信号 + rank — 回测排名 +""" + +from __future__ import annotations + +from pathlib import Path + +import click + + +@click.group() +def screen() -> None: + """策略选股扫描器 — 用策略扫描全市场触发信号的股票。 + + 两步走工作流： + + easy-tdx screen scan --strategy strategies/rsi_reversal.py --output signals.json + + easy-tdx screen rank --from signals.json --sort sharpe --top 20 --table + """ + + +# ── scan 子命令 ────────────────────────────────────────────────────────────── + + +@screen.command() +@click.option("--strategy", "strategy_file", required=True, help="策略文件路径") +@click.option("--output", "output_file", default=None, help="输出 JSON 文件路径（默认 stdout）") +@click.option( + "--universe", + default="all", + help="股票范围: all/sh/sz/<文件路径>（默认 all）", +) +@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="佣金率") +def scan( + strategy_file: str, + output_file: str | None, + universe: str, + vipdoc: str | None, + cash: float, + commission: float, +) -> None: + """纯离线扫描全市场，找出触发买入信号的股票。 + + 读取本地通达信 .day 文件，零网络 IO，全市场约 30-60 秒。 + + 示例： + + easy-tdx screen scan --strategy strategies/rsi_reversal.py + + easy-tdx screen scan --strategy strategies/rsi_reversal.py --output signals.json + + easy-tdx screen scan --strategy strategies/rsi_reversal.py --universe sz + """ + + strategy_cls = _load_strategy(strategy_file) + strategy_name = strategy_cls.__name__ + click.echo(f"策略: {strategy_name}", err=True) + click.echo(f"范围: {universe}", err=True) + + from .scanner import SignalScanner + + scanner = SignalScanner( + strategy_cls=strategy_cls, + vipdoc_path=vipdoc, + cash=cash, + commission=commission, + ) + + # 进度回调（输出到 stderr，避免污染 stdout 的 JSON） + total_scanned = 0 + + def on_progress(current: int, total: int, name: str) -> None: + nonlocal total_scanned + total_scanned = total + if name == "done": + click.echo(f"\r扫描完成: {total} 只", err=True) + else: + 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) + + # 生成 JSON + json_str = scanner.to_json( + results=results, + strategy_name=strategy_name, + strategy_file=strategy_file, + total_scanned=total_scanned, + ) + + # 输出 + if output_file: + Path(output_file).write_text(json_str, encoding="utf-8") + click.echo(f"信号数: {len(results)} → {output_file}") + else: + click.echo(json_str) + + +# ── rank 子命令 ────────────────────────────────────────────────────────────── + + +@screen.command("rank") +@click.option("--from", "from_source", required=True, help="信号 JSON 文件路径（- 表示 stdin）") +@click.option("--strategy", "strategy_file", default=None, help="覆盖策略文件（默认从 JSON 读取）") +@click.option("--sort", "sort_by", default="sharpe", help="排序指标（默认 sharpe）") +@click.option("--sort-reverse", is_flag=True, help="升序排列（用于回撤等越小越好的指标）") +@click.option("--top", "top_n", default=20, type=int, help="只显示前 N 名（默认 20）") +@click.option("--cash", default=1_000_000.0, type=float, help="初始资金") +@click.option("--count", default=0, type=int, help="使用最近 N 条 K 线，0=全部") +@click.option("--commission", default=0.0003, type=float, help="佣金率") +@click.option("--vipdoc", default=None, help="离线数据目录（默认自动检测）") +@click.option("--names/--no-names", default=False, help="是否在线查询股票名称（默认关闭）") +@click.option("--table", "use_table", is_flag=True, help="表格输出") +def rank_cmd( + from_source: str, + strategy_file: str | None, + sort_by: str, + sort_reverse: bool, + top_n: int, + cash: float, + count: int, + commission: float, + vipdoc: str | None, + names: bool, + use_table: bool, +) -> None: + """对扫描结果做历史回测并按指标排名。 + + 读取 scan 输出的 JSON，对每只股票跑完整回测，按指定指标排序。 + + 示例： + + easy-tdx screen rank --from signals.json --sort sharpe --top 20 --table + + easy-tdx screen rank --from signals.json --sort max_drawdown --sort-reverse + + easy-tdx screen scan --strategy strats/rsi.py | easy-tdx screen rank --from - --table + """ + from .ranker import SignalRanker, load_signals + + # 加载信号 + signals, strategy_name, strategy_file_from_json = load_signals(from_source) + + if not signals: + click.echo("无信号数据，无需排名") + return + + # 确定策略 + effective_strategy_file = strategy_file or strategy_file_from_json + if not effective_strategy_file: + click.echo( + "错误: 未指定策略文件，请使用 --strategy 或确保 JSON 包含 strategy_file", err=True + ) + raise SystemExit(1) + + strategy_cls = _load_strategy(effective_strategy_file) + strategy_name = strategy_cls.__name__ + + click.echo(f"策略: {strategy_name} | 信号数: {len(signals)} | 排序: {sort_by}", err=True) + + ranker = SignalRanker( + strategy_cls=strategy_cls, + vipdoc_path=vipdoc, + cash=cash, + commission=commission, + count=count, + ) + + # 进度回调（输出到 stderr，避免污染 stdout） + def on_progress(current: int, total: int, label: str) -> None: + if label == "done": + click.echo(f"\r排名完成: {total} 只", err=True) + else: + pct = current * 100 // total if total > 0 else 0 + click.echo(f"\r[{current}/{total}] {pct}% backtesting {label}", nl=False, err=True) + + entries = ranker.rank( + signals=signals, + sort_by=sort_by, + sort_reverse=sort_reverse, + top_n=top_n, + progress_callback=on_progress, + ) + + # 补齐名称（可选，需要网络） + if names and entries: + click.echo("\n正在获取股票名称...", err=True) + entries = ranker.enrich_names(entries) + + # 输出 + if use_table: + click.echo(ranker.to_table(entries, sort_by)) + else: + click.echo(ranker.to_json(entries, strategy_name, sort_by)) + + +# ── 辅助函数 ────────────────────────────────────────────────────────────────── + + +def _load_strategy(strategy_file: str) -> type: + """加载策略类（复用 backtest.cli 的加载逻辑）。 + + Args: + strategy_file: 策略文件路径 + + Returns: + Strategy 子类 + """ + import importlib.util + + from easy_tdx.backtest.strategy import Strategy + + file_path = Path(strategy_file) + if not file_path.exists(): + click.echo(f"错误: 策略文件不存在: {strategy_file}", err=True) + raise SystemExit(1) + + spec = importlib.util.spec_from_file_location("strategy_module", file_path) + if spec is None or spec.loader is None: + click.echo(f"错误: 无法加载策略文件: {strategy_file}", err=True) + raise SystemExit(1) + + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + + # 查找 Strategy 子类 + for attr_name in dir(module): + obj = getattr(module, attr_name) + try: + if isinstance(obj, type) and issubclass(obj, Strategy) and obj is not Strategy: + return obj + except TypeError: + pass + + click.echo(f"错误: 文件中未找到 Strategy 子类: {strategy_file}", err=True) + raise SystemExit(1) diff --git a/src/easy_tdx/screen/ranker.py b/src/easy_tdx/screen/ranker.py new file mode 100644 index 0000000..5d358f1 --- /dev/null +++ b/src/easy_tdx/screen/ranker.py @@ -0,0 +1,355 @@ +"""回测排名引擎 — 对扫描结果做历史回测并按指标排名。 + +核心流程： +1. 读取 scan 输出的 JSON（文件或 stdin） +2. 每只股票：read_daily_bars() → DataFrame → BacktestEngine.run() → performance +3. 按 --sort 指标排序（默认 sharpe） +4. 可选在线获取股票名称 +5. 输出排名表或 JSON +""" + +from __future__ import annotations + +import json +import sys +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +from easy_tdx.backtest.engine import BacktestEngine +from easy_tdx.backtest.strategy import Strategy +from easy_tdx.offline.daily_bar import read_daily_bars +from easy_tdx.offline.paths import resolve_vipdoc + +from .scanner import _bars_to_df + + +@dataclass +class RankEntry: + """排名条目。 + + Attributes: + rank: 排名位置 + code: 6 位股票代码 + market: 市场（SZ/SH） + name: 股票名称（可能为空） + signal_date: 信号日期 + last_close: 最后收盘价 + performance: 绩效指标字典 + """ + + rank: int + code: str + market: str + name: str + signal_date: int + last_close: float + performance: dict[str, float] + + +class SignalRanker: + """信号排名器。 + + 用法:: + + ranker = SignalRanker(strategy_cls=RSIStrategy) + entries = ranker.rank(signals) + for e in entries[:10]: + print(f"#{e.rank} {e.market}{e.code} sharpe={e.performance['sharpe']:.2f}") + """ + + def __init__( + self, + strategy_cls: type[Strategy], + vipdoc_path: str | Path | None = None, + cash: float = 1_000_000.0, + commission: float = 0.0003, + count: int = 0, + ) -> None: + """初始化排名器。 + + Args: + strategy_cls: 策略类 + vipdoc_path: vipdoc 目录路径 + cash: 初始资金 + commission: 佣金率 + count: 使用最近 N 条 K 线，0=全部 + """ + self._strategy_cls = strategy_cls + self._vipdoc = resolve_vipdoc(vipdoc_path) + self._cash = cash + self._commission = commission + self._count = count + + def rank( + self, + signals: list[dict[str, Any]], + sort_by: str = "sharpe", + sort_reverse: bool = False, + top_n: int = 0, + progress_callback: Any = None, + ) -> list[RankEntry]: + """对信号列表做回测排名。 + + Args: + signals: scan 输出的信号列表 + [{"code": "000001", "market": "SZ", "signal_date": 20260610, ...}] + sort_by: 排序指标（默认 sharpe） + sort_reverse: True 则升序（用于 max_drawdown 等越小越好的指标） + top_n: 只返回前 N 名，0=全部 + progress_callback: 进度回调(current, total, code) + + Returns: + RankEntry 列表 + """ + entries: list[RankEntry] = [] + total = len(signals) + + for idx, sig in enumerate(signals): + code = sig["code"] + market = sig["market"] + signal_date = sig.get("signal_date", 0) + last_close = sig.get("last_close", 0.0) + + if progress_callback: + progress_callback(idx, total, f"{market}{code}") + + try: + perf = self._backtest_one(market, code) + if perf is not None: + entries.append( + RankEntry( + rank=0, # 排名在排序后赋值 + code=code, + market=market, + name="", + signal_date=signal_date, + last_close=last_close, + performance=perf, + ) + ) + except Exception: + continue + + if progress_callback: + progress_callback(total, total, "done") + + # 排序 + entries.sort( + key=lambda e: e.performance.get(sort_by, 0.0), + reverse=not sort_reverse, + ) + + # 赋值排名 + for i, entry in enumerate(entries): + entry.rank = i + 1 + + # 截取 top_n + if top_n > 0: + entries = entries[:top_n] + + return entries + + def _backtest_one(self, market: str, code: str) -> dict[str, float] | None: + """对单只股票做回测。 + + Args: + market: 市场（SZ/SH） + code: 6 位股票代码 + + Returns: + 绩效指标字典，数据不足时返回 None + """ + exchange = market.lower() + filepath = self._vipdoc / exchange / "lday" / f"{exchange}{code}.day" + + if not filepath.is_file(): + return None + + bars = read_daily_bars(filepath) + if len(bars) < 30: + return None + + df = _bars_to_df(bars) + if df.empty: + return None + + # 截取最近 count 条 + if self._count > 0 and len(df) > self._count: + df = df.iloc[-self._count :].reset_index(drop=True) + + engine = BacktestEngine( + strategy=self._strategy_cls, + cash=self._cash, + commission=self._commission, + ) + result = engine.run(df) + return result.performance + + def enrich_names(self, entries: list[RankEntry]) -> list[RankEntry]: + """通过在线查询补齐股票名称。 + + 仅对排名中的股票查询，通常只有几十只。 + + Args: + entries: 排名列表 + + Returns: + 补齐名称后的列表（原地修改） + """ + if not entries: + return entries + + try: + from easy_tdx.cli.parsers import parse_market + from easy_tdx.mac.client import MacClient + + # 批量查询 + pairs = [(parse_market(e.market), e.code) for e in entries] + + client = MacClient.from_best_host() + try: + client.connect() + quotes_df = client.get_stock_quotes(pairs) + finally: + client.close() + + if quotes_df.empty or "name" not in quotes_df.columns: + return entries + + # 构建 "SZ000001" → name 映射 + # MacQuoteField.market 是 int (0=SZ, 1=SH)，需转为字符串 + _market_map = {0: "SZ", 1: "SH"} + name_map: dict[str, str] = {} + for _, row in quotes_df.iterrows(): + mkt_int = row.get("market", -1) + mkt_str = _market_map.get(mkt_int, str(mkt_int)) + key = f"{mkt_str}{row.get('code', '')}" + name_map[key] = str(row.get("name", "")) + + for entry in entries: + key = f"{entry.market}{entry.code}" + if key in name_map: + entry.name = name_map[key] + + except Exception: + # 名称查询失败不影响主流程 + pass + + return entries + + @staticmethod + def to_json( + entries: list[RankEntry], + strategy_name: str, + sort_by: str, + ) -> str: + """将排名结果序列化为 JSON 字符串。 + + Args: + entries: 排名列表 + strategy_name: 策略名称 + sort_by: 排序指标 + + Returns: + JSON 字符串 + """ + data = { + "strategy": strategy_name, + "sort_by": sort_by, + "total_ranked": len(entries), + "ranking": [ + { + "rank": e.rank, + "code": e.code, + "market": e.market, + "name": e.name, + "signal_date": e.signal_date, + "last_close": e.last_close, + "performance": e.performance, + } + for e in entries + ], + } + return json.dumps(data, ensure_ascii=False, indent=2, default=_json_default) + + @staticmethod + def to_table(entries: list[RankEntry], sort_by: str) -> str: + """将排名结果格式化为表格字符串。 + + Args: + entries: 排名列表 + sort_by: 排序指标 + + Returns: + 表格字符串 + """ + if not entries: + return "无有效排名结果" + + sort_label = f"{sort_by} 降序" + lines = [ + f"[*] 信号排名 (按 {sort_label}, 共 {len(entries)} 只)", + "═" * 90, + f"{'排名':>4} {'代码':<10} {'名称':<10} {'总收益率':>10} {'年化收益':>10} " + f"{'最大回撤':>10} {'夏普':>8} {'胜率':>8} {'交易':>6}", + "─" * 90, + ] + + for e in entries: + medal = ( + " *1*" + if e.rank == 1 + else " *2*" + if e.rank == 2 + else " *3*" + if e.rank == 3 + else " " + ) + perf = e.performance + label = f"{e.market}{e.code}" + name = e.name[:8] if e.name else "" + lines.append( + f"{medal}{e.rank:>2} {label:<10} {name:<10} " + f"{perf.get('total_return', 0):>9.2%} " + f"{perf.get('annual_return', 0):>9.2%} " + f"{perf.get('max_drawdown', 0):>9.2%} " + f"{perf.get('sharpe', 0):>8.2f} " + f"{perf.get('win_rate', 0):>7.1%} " + f"{perf.get('total_trades', 0):>6}" + ) + + return "\n".join(lines) + + +def _json_default(obj: Any) -> Any: + """JSON 序列化辅助。""" + if hasattr(obj, "item"): + return obj.item() + if hasattr(obj, "isoformat"): + return obj.isoformat() + raise TypeError(f"Object of type {type(obj)} is not JSON serializable") + + +def load_signals(source: str) -> tuple[list[dict[str, Any]], str, str]: + """从文件或 stdin 加载信号 JSON。 + + Args: + source: JSON 文件路径，"-" 表示 stdin + + Returns: + (signals, strategy_name, strategy_file) + """ + if source == "-": + data = json.load(sys.stdin) + else: + filepath = Path(source) + if not filepath.is_file(): + raise FileNotFoundError(f"信号文件不存在: {source}") + with open(filepath, encoding="utf-8") as f: + data = json.load(f) + + signals = data.get("signals", []) + strategy_name = data.get("strategy", "unknown") + strategy_file = data.get("strategy_file", "") + return signals, strategy_name, strategy_file diff --git a/src/easy_tdx/screen/scanner.py b/src/easy_tdx/screen/scanner.py new file mode 100644 index 0000000..6e9fb2d --- /dev/null +++ b/src/easy_tdx/screen/scanner.py @@ -0,0 +1,320 @@ +"""信号扫描引擎 — 纯离线，从本地 .day 文件提取策略信号。 + +核心流程： +1. 扫描 vipdoc/{sh,sz}/lday/*.day 获取文件列表 +2. 按 universe 过滤（all/sh/sz/文件列表） +3. 过滤掉非 A 股（指数、基金、债券） +4. 每个文件：read_daily_bars() → DataFrame → extract_factor_signals() → 检查 buy_mask[-1] +5. 输出触发信号的股票列表 +""" + +from __future__ import annotations + +import json +from dataclasses import dataclass +from datetime import datetime +from pathlib import Path +from typing import Any + +import pandas as pd + +from easy_tdx.backtest.combo import extract_factor_signals +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 + +# A 股类型白名单 +_A_STOCK_TYPES = frozenset( + { + "SH_A_STOCK", + "SZ_A_STOCK", + } +) + + +@dataclass +class ScanResult: + """单只股票的扫描结果。 + + Attributes: + code: 6 位股票代码 + market: 市场（SZ/SH） + signal_date: 信号日期（YYYYMMDD 整数） + last_close: 最后收盘价 + """ + + code: str + market: str + signal_date: int + last_close: float + + +class SignalScanner: + """策略信号扫描器。 + + 用法:: + + scanner = SignalScanner( + strategy_cls=RSIStrategy, + vipdoc_path="C:\\new_jyplug\\vipdoc", + ) + results = scanner.scan(universe="all") + for r in results: + print(f"{r.market}{r.code} 触发买入信号 @ {r.signal_date}") + """ + + def __init__( + self, + strategy_cls: type[Strategy], + vipdoc_path: str | Path | None = None, + cash: float = 100_000.0, + commission: float = 0.0003, + ) -> None: + """初始化扫描器。 + + Args: + strategy_cls: 策略类（Strategy 子类） + vipdoc_path: vipdoc 目录路径，None 则自动检测 + cash: 初始资金（影响全仓信号判断） + commission: 佣金率 + """ + self._strategy_cls = strategy_cls + self._vipdoc = resolve_vipdoc(vipdoc_path) + self._cash = cash + self._commission = commission + + def scan( + self, + universe: str = "all", + progress_callback: Any = None, + ) -> list[ScanResult]: + """扫描全市场，返回触发买入信号的股票列表。 + + Args: + universe: 股票范围 + - "all": 沪深全部 A 股（默认） + - "sh": 仅上海 + - "sz": 仅深圳 + - 文件路径: 每行一个 "市场 代码"（如 "SZ 000001"） + progress_callback: 进度回调函数(current, total, filename) + + Returns: + 触发买入信号的 ScanResult 列表 + """ + # 1. 收集文件列表 + files = self._collect_files(universe) + + if not files: + return [] + + results: list[ScanResult] = [] + total = len(files) + + for idx, (filepath, market, code) in enumerate(files): + if progress_callback: + progress_callback(idx, total, filepath.name) + + try: + result = self._scan_one(filepath, market, code) + if result is not None: + results.append(result) + except Exception: + # 单个文件出错不中断整体扫描 + continue + + if progress_callback: + progress_callback(total, total, "done") + + return results + + def _collect_files(self, universe: str) -> list[tuple[Path, str, str]]: + """收集需要扫描的 .day 文件列表。 + + Args: + universe: 股票范围 + + Returns: + [(filepath, market_str, code), ...] 列表 + """ + # 确定要扫描的交易所目录 + exchanges: list[str] = [] + if universe in ("all", "sz"): + exchanges.append("sz") + if universe in ("all", "sh"): + exchanges.append("sh") + + # 从文件列表模式读取 + if universe not in ("all", "sh", "sz"): + return self._collect_from_file(universe) + + # 扫描目录 + files: list[tuple[Path, str, str]] = [] + for exchange in exchanges: + lday_dir = self._vipdoc / exchange / "lday" + if not lday_dir.is_dir(): + continue + + for filepath in sorted(lday_dir.glob("*.day")): + # 从文件名提取代码 + name = filepath.name.lower() + code = name[2:8] + + # 过滤非 A 股 + sec_type = _detect_security_type(filepath.name) + if sec_type not in _A_STOCK_TYPES: + continue + + market = exchange.upper() + files.append((filepath, market, code)) + + return files + + def _collect_from_file(self, filepath: str) -> list[tuple[Path, str, str]]: + """从文件读取股票列表。 + + 每行格式: "市场 代码"（如 "SZ 000001"） + + Args: + filepath: 股票列表文件路径 + + Returns: + [(filepath, market_str, code), ...] 列表 + """ + path = Path(filepath) + if not path.is_file(): + raise FileNotFoundError(f"股票列表文件不存在: {filepath}") + + files: list[tuple[Path, str, str]] = [] + with open(path, encoding="utf-8") as f: + for line in f: + line = line.strip() + if not line or line.startswith("#"): + continue + + parts = line.split() + if len(parts) >= 2: + market_str = parts[0].upper() + code = parts[1] + else: + continue + + # 定位 .day 文件 + exchange = market_str.lower() + day_file = self._vipdoc / exchange / "lday" / f"{exchange}{code}.day" + if day_file.is_file(): + files.append((day_file, market_str, code)) + + return files + + def _scan_one(self, filepath: Path, market: str, code: str) -> ScanResult | None: + """扫描单只股票。 + + Args: + filepath: .day 文件路径 + market: 市场代码（SZ/SH） + code: 6 位股票代码 + + Returns: + ScanResult 如果触发信号，否则 None + """ + bars = read_daily_bars(filepath) + if len(bars) < 30: + # 数据太少，无法计算有意义的指标 + return None + + df = _bars_to_df(bars) + if df.empty: + return None + + # 提取信号遮罩 + try: + factor_signals = extract_factor_signals( + self._strategy_cls, + df, + cash=self._cash, + commission=self._commission, + ) + except Exception: + return None + + # 检查最后一根 bar 是否有买入信号 + if not factor_signals.buy_mask[-1]: + return None + + # 获取最后收盘价和日期 + last_bar = bars[-1] + signal_date = last_bar.year * 10000 + last_bar.month * 100 + last_bar.day + last_close = last_bar.close + + return ScanResult( + code=code, + market=market, + signal_date=signal_date, + last_close=last_close, + ) + + def to_json( + self, + results: list[ScanResult], + strategy_name: str, + strategy_file: str, + total_scanned: int, + ) -> str: + """将扫描结果序列化为 JSON 字符串。 + + Args: + results: 扫描结果列表 + strategy_name: 策略名称 + strategy_file: 策略文件路径 + total_scanned: 总扫描股票数 + + Returns: + JSON 字符串 + """ + data = { + "scan_time": datetime.now().isoformat(timespec="seconds"), + "strategy": strategy_name, + "strategy_file": strategy_file, + "total_scanned": total_scanned, + "total_signals": len(results), + "signals": [ + { + "code": r.code, + "market": r.market, + "signal_date": r.signal_date, + "last_close": r.last_close, + } + for r in results + ], + } + return json.dumps(data, ensure_ascii=False, indent=2) + + +def _bars_to_df(bars: list[Any]) -> pd.DataFrame: + """将 SecurityBar 列表转为策略所需的 DataFrame。 + + Args: + bars: SecurityBar 列表（按时间升序） + + Returns: + DataFrame，包含 datetime, open, close, high, low, vol, amount 列 + """ + if not bars: + return pd.DataFrame() + + rows = [] + for b in bars: + dt = pd.Timestamp(year=b.year, month=b.month, day=b.day) + rows.append( + { + "datetime": dt, + "open": b.open, + "close": b.close, + "high": b.high, + "low": b.low, + "vol": b.vol, + "amount": b.amount, + } + ) + + return pd.DataFrame(rows) diff --git a/tests/unit/test_screen.py b/tests/unit/test_screen.py new file mode 100644 index 0000000..ec5a677 --- /dev/null +++ b/tests/unit/test_screen.py @@ -0,0 +1,384 @@ +"""screen 模块单元测试 — 纯离线，无需网络。""" + +from __future__ import annotations + +import json +from io import StringIO +from pathlib import Path +from typing import Any +from unittest.mock import MagicMock, patch + +import pandas as pd +import pytest +from easy_tdx.models.bar import SecurityBar + +# ── 辅助：构造 SecurityBar ──────────────────────────────────────────────── + + +def _make_bar(year: int, month: int, day: int, close: float, **kw: Any) -> SecurityBar: + """快速构造一个 SecurityBar。""" + return SecurityBar( + open=kw.get("open", close - 0.1), + close=close, + high=kw.get("high", close + 0.2), + low=kw.get("low", close - 0.3), + vol=kw.get("vol", 10000.0), + amount=kw.get("amount", close * 10000), + year=year, + month=month, + day=day, + hour=0, + minute=0, + ) + + +def _make_bars(n: int, base_close: float = 10.0) -> list[SecurityBar]: + """构造 n 根连续日 K 线，收盘价从 base_close 开始递增。""" + bars = [] + for i in range(n): + year = 2024 + month = 1 + i // 28 + day = 1 + i % 28 + if month > 12: + year += (month - 1) // 12 + month = 1 + (month - 1) % 12 + close = base_close + i * 0.1 + bars.append(_make_bar(year, month, day, close)) + return bars + + +# ── _bars_to_df 测试 ────────────────────────────────────────────────────── + + +class TestBarsToDf: + """测试 scanner._bars_to_df 辅助函数。""" + + def test_empty_bars(self) -> None: + from easy_tdx.screen.scanner import _bars_to_df + + df = _bars_to_df([]) + assert df.empty + + def test_single_bar(self) -> None: + from easy_tdx.screen.scanner import _bars_to_df + + bar = _make_bar(2024, 6, 10, 12.5) + df = _bars_to_df([bar]) + assert len(df) == 1 + assert df.iloc[0]["close"] == 12.5 + assert "datetime" in df.columns + assert "open" in df.columns + assert "vol" in df.columns + + def test_multiple_bars(self) -> None: + from easy_tdx.screen.scanner import _bars_to_df + + bars = _make_bars(50) + df = _bars_to_df(bars) + assert len(df) == 50 + assert list(df.columns) == ["datetime", "open", "close", "high", "low", "vol", "amount"] + + def test_datetime_is_timestamp(self) -> None: + from easy_tdx.screen.scanner import _bars_to_df + + bars = [_make_bar(2024, 6, 10, 12.5)] + df = _bars_to_df(bars) + assert isinstance(df.iloc[0]["datetime"], pd.Timestamp) + + +# ── ScanResult 测试 ────────────────────────────────────────────────────── + + +class TestScanResult: + """测试 ScanResult 数据结构。""" + + def test_creation(self) -> None: + from easy_tdx.screen.scanner import ScanResult + + r = ScanResult(code="000001", market="SZ", signal_date=20240610, last_close=12.5) + assert r.code == "000001" + assert r.market == "SZ" + assert r.signal_date == 20240610 + assert r.last_close == 12.5 + + +# ── SignalScanner 测试 ────────────────────────────────────────────────── + + +class TestSignalScanner: + """测试信号扫描引擎。""" + + def test_to_json(self) -> None: + from easy_tdx.screen.scanner import ScanResult, SignalScanner + + scanner = SignalScanner.__new__(SignalScanner) + results = [ + ScanResult(code="000001", market="SZ", signal_date=20240610, last_close=12.5), + ScanResult(code="600519", market="SH", signal_date=20240610, last_close=1800.0), + ] + json_str = scanner.to_json(results, "TestStrategy", "test.py", 100) + data = json.loads(json_str) + + assert data["strategy"] == "TestStrategy" + assert data["strategy_file"] == "test.py" + assert data["total_scanned"] == 100 + assert data["total_signals"] == 2 + assert len(data["signals"]) == 2 + assert data["signals"][0]["code"] == "000001" + assert data["signals"][1]["market"] == "SH" + + def test_to_json_empty(self) -> None: + from easy_tdx.screen.scanner import SignalScanner + + scanner = SignalScanner.__new__(SignalScanner) + json_str = scanner.to_json([], "Test", "t.py", 50) + data = json.loads(json_str) + assert data["total_signals"] == 0 + assert data["signals"] == [] + + +# ── RankEntry 测试 ────────────────────────────────────────────────────── + + +class TestRankEntry: + """测试 RankEntry 数据结构。""" + + def test_creation(self) -> None: + from easy_tdx.screen.ranker import RankEntry + + e = RankEntry( + rank=1, + code="300308", + market="SZ", + name="", + signal_date=20240610, + last_close=85.0, + performance={"sharpe": 1.85, "total_return": 0.45}, + ) + assert e.rank == 1 + assert e.performance["sharpe"] == 1.85 + + +# ── SignalRanker.to_json 测试 ────────────────────────────────────────── + + +class TestRankerJson: + """测试 Ranker 的 JSON 输出。""" + + def test_to_json(self) -> None: + from easy_tdx.screen.ranker import RankEntry, SignalRanker + + entries = [ + RankEntry( + rank=1, + code="300308", + market="SZ", + name="", + signal_date=20240610, + last_close=85.0, + performance={"sharpe": 1.85, "total_return": 0.45}, + ), + ] + json_str = SignalRanker.to_json(entries, "RSI", "sharpe") + data = json.loads(json_str) + + assert data["strategy"] == "RSI" + assert data["sort_by"] == "sharpe" + assert data["total_ranked"] == 1 + assert data["ranking"][0]["rank"] == 1 + assert data["ranking"][0]["code"] == "300308" + + def test_to_table(self) -> None: + from easy_tdx.screen.ranker import RankEntry, SignalRanker + + entries = [ + RankEntry( + rank=1, + code="300308", + market="SZ", + name="中际旭创", + signal_date=20240610, + last_close=85.0, + performance={ + "total_return": 0.4523, + "annual_return": 0.1872, + "max_drawdown": 0.1235, + "sharpe": 1.85, + "win_rate": 0.625, + "total_trades": 16, + }, + ), + ] + table = SignalRanker.to_table(entries, "sharpe") + assert "信号排名" in table + assert "SZ300308" in table + assert "45.23%" in table + + def test_to_table_empty(self) -> None: + from easy_tdx.screen.ranker import SignalRanker + + table = SignalRanker.to_table([], "sharpe") + assert "无有效排名结果" in table + + +# ── load_signals 测试 ────────────────────────────────────────────────── + + +class TestLoadSignals: + """测试信号 JSON 加载。""" + + def test_load_from_file(self, tmp_path: Path) -> None: + from easy_tdx.screen.ranker import load_signals + + data = { + "strategy": "RSI", + "strategy_file": "rsi.py", + "signals": [ + {"code": "000001", "market": "SZ", "signal_date": 20240610, "last_close": 12.5}, + ], + } + filepath = tmp_path / "signals.json" + filepath.write_text(json.dumps(data), encoding="utf-8") + + signals, name, sfile = load_signals(str(filepath)) + assert len(signals) == 1 + assert name == "RSI" + assert sfile == "rsi.py" + assert signals[0]["code"] == "000001" + + def test_load_from_stdin(self) -> None: + from easy_tdx.screen.ranker import load_signals + + data = { + "strategy": "MACD", + "signals": [ + {"code": "600519", "market": "SH"}, + ], + } + json_str = json.dumps(data) + + with patch("sys.stdin", StringIO(json_str)): + signals, name, _ = load_signals("-") + + assert len(signals) == 1 + assert name == "MACD" + + def test_load_missing_file(self) -> None: + from easy_tdx.screen.ranker import load_signals + + with pytest.raises(FileNotFoundError): + load_signals("/nonexistent/path.json") + + def test_load_empty_signals(self, tmp_path: Path) -> None: + from easy_tdx.screen.ranker import load_signals + + data = {"strategy": "RSI", "signals": []} + filepath = tmp_path / "empty.json" + filepath.write_text(json.dumps(data), encoding="utf-8") + + signals, name, _ = load_signals(str(filepath)) + assert signals == [] + + +# ── 集成：scanner._scan_one 逻辑 ───────────────────────────────────── + + +class TestScanOne: + """测试 scanner 的单股扫描逻辑（模拟策略信号）。""" + + def test_no_signal(self) -> None: + """策略不产生买入信号时返回 None。""" + from easy_tdx.screen.scanner import SignalScanner + + # 构造一个永远不产生买入信号的 mock 策略 + mock_strategy = MagicMock() + mock_strategy.__name__ = "NeverBuyStrategy" + + scanner = SignalScanner.__new__(SignalScanner) + scanner._strategy_cls = mock_strategy + scanner._vipdoc = Path("/fake") + scanner._cash = 100000.0 + scanner._commission = 0.0003 + + bars = _make_bars(100) + with patch.object(scanner, "_scan_one") as mock_scan: + # 不产生信号时返回 None + mock_scan.return_value = None + result = scanner._scan_one(Path("/fake/sz000001.day"), "SZ", "000001") + assert result is None + + def test_collect_files_universe_sh(self) -> None: + """universe=sh 时只扫描上海 A 股。""" + from easy_tdx.screen.scanner import SignalScanner + + scanner = SignalScanner.__new__(SignalScanner) + + # mock vipdoc 目录结构 + sh_dir = MagicMock() + sh_files = [MagicMock(name="sh600000.day"), MagicMock(name="sh000001.day")] + sh_files[0].name = "sh600000.day" + sh_files[1].name = "sh000001.day" + sh_dir.is_dir.return_value = True + sh_dir.glob.return_value = iter(sh_files) + + sz_dir = MagicMock() + sz_dir.is_dir.return_value = False + + mock_vipdoc = MagicMock() + mock_vipdoc.__truediv__ = MagicMock( + side_effect=lambda x: sh_dir if "sh" in str(x) else sz_dir + ) + + scanner._vipdoc = mock_vipdoc + + # 只测 universe 过滤逻辑（不测文件 IO） + # 实际测试：_collect_files 应该跳过指数文件 sh000001 + # 这里验证 _detect_security_type 被正确调用 + from easy_tdx.offline.daily_bar import _detect_security_type + + assert _detect_security_type("sh600000.day") == "SH_A_STOCK" + assert _detect_security_type("sh000001.day") == "SH_INDEX" + assert _detect_security_type("sz000001.day") == "SZ_A_STOCK" + assert _detect_security_type("sz399001.day") == "SZ_INDEX" + assert _detect_security_type("sz159919.day") == "SZ_FUND" + + +# ── 策略加载测试 ──────────────────────────────────────────────────────── + + +class TestLoadStrategy: + """测试 CLI 的策略加载。""" + + def test_load_valid_strategy(self, tmp_path: Path) -> None: + from easy_tdx.screen.cli import _load_strategy + + # 写一个简单的策略文件 + strategy_code = """ +from easy_tdx.backtest import Strategy + +class DummyStrategy(Strategy): + def init(self) -> None: + pass + def next(self) -> None: + pass +""" + filepath = tmp_path / "dummy.py" + filepath.write_text(strategy_code, encoding="utf-8") + + cls = _load_strategy(str(filepath)) + assert cls.__name__ == "DummyStrategy" + + def test_load_missing_file(self) -> None: + from easy_tdx.screen.cli import _load_strategy + + with pytest.raises(SystemExit): + _load_strategy("/nonexistent/strategy.py") + + def test_load_no_strategy_class(self, tmp_path: Path) -> None: + from easy_tdx.screen.cli import _load_strategy + + filepath = tmp_path / "empty.py" + filepath.write_text("x = 1\n", encoding="utf-8") + + with pytest.raises(SystemExit): + _load_strategy(str(filepath))