aeroxw/easy-tdx
1
0
Fork 0
mirror of https://ghfast.top/https://github.com/aeroxw/easy-tdx.git synced 2026-09-12 14:34:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: multi-factor combo backtest engine (v1.9.0)

Browse Source 
- Add backtest/combo.py: CombinationRunner, extract_factor_signals, combine_masks
- Signal merge modes: AND / OR / MAJORITY (majority default)
- CLI: --combo-strategies and --combo-mode for easy-tdx backtest
- run_all_strategies.py: --combo 2 --combo 3 auto-screen best combos
- Fix MyTT MFI/CR divide-by-zero RuntimeWarning
- 14 new unit tests, 328 total passing
This commit is contained in:
Justin Gu
2026-06-10 01:37:28 +08:00
parent 7f1bc645c2
commit 1e99feb7c2
14 changed files with 1695 additions and 578 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.omc/project-memory.json
+181 -295
View File
@@ -1,7 +1,7 @@
{
"version": "1.0.0",
"lastScanned": 1779337271902,
"projectRoot": "D:\\python\\xmtdx",
"lastScanned": 1780948623576,
"projectRoot": "D:\\python\\easy_tdx",
"techStack": {
"languages": [
{
@@ -25,7 +25,7 @@
},
"build": {
"buildCommand": null,
"testCommand": "python -m ruff check src/ tests/ 2>&1 && python -m pytest tests/unit/ -q 2>&1",
"testCommand": "python -m pytest tests/unit/ -q --tb=line 2>&1 | tail -5",
"lintCommand": "ruff check",
"devCommand": null,
"scripts": {}
@@ -40,6 +40,8 @@
"isMonorepo": false,
"workspaces": [],
"mainDirectories": [
"docs",
"examples",
"scripts",
"src",
"tests"
@@ -51,13 +53,44 @@
},
"customNotes": [],
"directoryMap": {
"dist": {
"path": "dist",
"purpose": "Distribution/build output",
"fileCount": 2,
"lastAccessed": 1780948623545,
"keyFiles": [
"easy_tdx-1.0.0-py3-none-any.whl",
"easy_tdx-1.0.0.tar.gz"
]
},
"docs": {
"path": "docs",
"purpose": "Documentation",
"fileCount": 8,
"lastAccessed": 1780948623545,
"keyFiles": [
"api_reference.md",
"conf.py",
"field_mapping.md",
"index.md",
"indicator-bias-signal.md"
]
},
"examples": {
"path": "examples",
"purpose": "Example code",
"fileCount": 0,
"lastAccessed": 1780948623546,
"keyFiles": []
},
"scripts": {
"path": "scripts",
"purpose": "Build/utility scripts",
"fileCount": 3,
"lastAccessed": 1779337271876,
"fileCount": 4,
"lastAccessed": 1780948623547,
"keyFiles": [
"probe_unknowns.py",
"ruff_hook.py",
"verify_fixes_20260415.py",
"__init__.py"
]
@@ -66,329 +99,182 @@
"path": "src",
"purpose": "Source code",
"fileCount": 0,
"lastAccessed": 1779337271876,
"lastAccessed": 1780948623548,
"keyFiles": []
},
"tests": {
"path": "tests",
"purpose": "Test files",
"fileCount": 1,
"lastAccessed": 1779337271877,
"lastAccessed": 1780948623549,
"keyFiles": [
"__init__.py"
]
},
"venv": {
"path": "venv",
"purpose": null,
"fileCount": 2,
"lastAccessed": 1780948623550,
"keyFiles": [
"pyvenv.cfg"
]
},
"tests\\fixtures": {
"path": "tests\\fixtures",
"purpose": "Test fixtures",
"fileCount": 24,
"lastAccessed": 1779337271879,
"lastAccessed": 1780948623554,
"keyFiles": [
"company_info_category.hex",
"company_info_category.json",
"company_info_content.hex"
]
},
"venv\\Lib": {
"path": "venv\\Lib",
"purpose": "Library code",
"fileCount": 0,
"lastAccessed": 1780948623555,
"keyFiles": []
},
"venv\\Scripts": {
"path": "venv\\Scripts",
"purpose": "Build/utility scripts",
"fileCount": 25,
"lastAccessed": 1780948623555,
"keyFiles": [
"activate",
"activate.bat",
"activate.fish"
]
}
},
"hotPaths": [
{
"path": "src\\easy_tdx\\client.py",
"accessCount": 116,
"lastAccessed": 1779392745160,
"path": "src\\easy_tdx\\backtest\\cli.py",
"accessCount": 14,
"lastAccessed": 1781024440572,
"type": "file"
},
{
"path": "run_all_strategies.py",
"accessCount": 13,
"lastAccessed": 1781024439395,
"type": "file"
},
{
"path": "src\\easy_tdx\\MyTT.py",
"accessCount": 9,
"lastAccessed": 1781025723850,
"type": "file"
},
{
"path": "README.md",
"accessCount": 16,
"lastAccessed": 1779394894399,
"type": "file"
},
{
"path": "src\\easy_tdx\\_df.py",
"accessCount": 12,
"lastAccessed": 1779391749202,
"type": "file"
},
{
"path": "tests\\unit\\test_a_share_extensions.py",
"accessCount": 9,
"lastAccessed": 1779392575071,
"type": "file"
},
{
"path": "tests\\unit\\test_block_info.py",
"accessCount": 6,
"lastAccessed": 1779387538331,
"type": "file"
},
{
"path": "examples\\05_transaction\\transaction_data.py",
"accessCount": 6,
"lastAccessed": 1779390229059,
"type": "file"
},
{
"path": "examples\\05_transaction\\history_transaction.py",
"accessCount": 6,
"lastAccessed": 1779390229831,
"type": "file"
},
{
"path": "src\\easy_tdx\\models\\timeseries.py",
"accessCount": 6,
"lastAccessed": 1779391466426,
"type": "file"
},
{
"path": "examples\\04_minute\\minute_time_data.py",
"accessCount": 4,
"lastAccessed": 1779392745091,
"lastAccessed": 1781026610801,
"type": "file"
},
{
"path": "src\\easy_tdx\\backtest\\combo.py",
"accessCount": 3,
"lastAccessed": 1781023715594,
"type": "file"
},
{
"path": "src\\easy_tdx\\backtest\\__init__.py",
"accessCount": 3,
"lastAccessed": 1781023799015,
"type": "file"
},
{
"path": "tests\\unit\\test_backtest_combo.py",
"accessCount": 2,
"lastAccessed": 1781023758085,
"type": "file"
},
{
"path": "src\\easy_tdx\\backtest\\strategy.py",
"accessCount": 1,
"lastAccessed": 1781021955703,
"type": "file"
},
{
"path": "src\\easy_tdx\\backtest\\engine.py",
"accessCount": 1,
"lastAccessed": 1781021956588,
"type": "file"
},
{
"path": "src\\easy_tdx\\backtest\\types.py",
"accessCount": 1,
"lastAccessed": 1781021957067,
"type": "file"
},
{
"path": "src\\easy_tdx\\backtest\\dsl.py",
"accessCount": 1,
"lastAccessed": 1781021962044,
"type": "file"
},
{
"path": "strategies\\ma_cross.py",
"accessCount": 1,
"lastAccessed": 1781021963164,
"type": "file"
},
{
"path": "strategies\\macd_cross.py",
"accessCount": 1,
"lastAccessed": 1781021963925,
"type": "file"
},
{
"path": "strategies\\rsi_reversal.py",
"accessCount": 1,
"lastAccessed": 1781021964466,
"type": "file"
},
{
"path": "strategies\\bollinger_breakout.py",
"accessCount": 1,
"lastAccessed": 1781021964950,
"type": "file"
},
{
"path": "src\\easy_tdx\\backtest\\performance.py",
"accessCount": 1,
"lastAccessed": 1781021974264,
"type": "file"
},
{
"path": "strategies\\turtle_breakout.py",
"accessCount": 1,
"lastAccessed": 1781021982151,
"type": "file"
},
{
"path": "C:\\Users\\毛利哥\\.claude\\plans\\2-3-deep-melody.md",
"accessCount": 1,
"lastAccessed": 1781022126312,
"type": "file"
},
{
"path": "tests\\unit\\test_backtest_strategy.py",
"accessCount": 1,
"lastAccessed": 1781023334642,
"type": "file"
},
{
"path": "tests\\unit\\test_backtest_engine.py",
"accessCount": 1,
"lastAccessed": 1781023335435,
"type": "file"
},
{
"path": "pyproject.toml",
"accessCount": 4,
"lastAccessed": 1779394669812,
"type": "file"
},
{
"path": "src",
"accessCount": 4,
"lastAccessed": 1779394669916,
"type": "directory"
},
{
"path": "scripts\\ruff_hook.py",
"accessCount": 3,
"lastAccessed": 1779337842663,
"type": "file"
},
{
"path": "src\\xmtdx\\transport\\sync.py",
"accessCount": 3,
"lastAccessed": 1779338445930,
"type": "file"
},
{
"path": "CLAUDE.md",
"accessCount": 3,
"lastAccessed": 1779386454500,
"type": "file"
},
{
"path": "src/easy_tdx/client.py",
"accessCount": 3,
"lastAccessed": 1779387035554,
"type": "directory"
},
{
"path": "examples\\03_kline\\security_bars.py",
"accessCount": 3,
"lastAccessed": 1779389362430,
"type": "file"
},
{
"path": "examples\\03_kline\\index_bars.py",
"accessCount": 3,
"lastAccessed": 1779389362522,
"type": "file"
},
{
"path": "examples\\06_finance\\xdxr_info.py",
"accessCount": 3,
"lastAccessed": 1779389518814,
"type": "file"
},
{
"path": "examples\\08_fund_flow\\history_fund_flow.py",
"accessCount": 3,
"lastAccessed": 1779389518888,
"type": "file"
},
{
"path": "examples\\02_market_info\\security_quotes.py",
"accessCount": 2,
"lastAccessed": 1779386629339,
"type": "file"
},
{
"path": "examples\\02_market_info\\security_list.py",
"accessCount": 2,
"lastAccessed": 1779386630075,
"type": "file"
},
{
"path": "examples\\02_market_info\\security_list_all.py",
"accessCount": 2,
"lastAccessed": 1779386630813,
"type": "file"
},
{
"path": "examples\\02_market_info\\market_stat.py",
"accessCount": 2,
"lastAccessed": 1779386631522,
"type": "file"
},
{
"path": "examples\\04_minute\\history_minute_data.py",
"accessCount": 2,
"lastAccessed": 1779386654282,
"type": "file"
},
{
"path": "examples\\06_finance\\finance_info.py",
"accessCount": 2,
"lastAccessed": 1779386691184,
"type": "file"
},
{
"path": "examples\\06_finance\\price_limits.py",
"accessCount": 2,
"lastAccessed": 1779386691927,
"type": "file"
},
{
"path": "examples\\06_finance\\company_info.py",
"accessCount": 2,
"lastAccessed": 1779386692682,
"type": "file"
},
{
"path": "examples\\07_block\\block_info.py",
"accessCount": 2,
"lastAccessed": 1779386693419,
"type": "file"
},
{
"path": "examples\\08_fund_flow\\fund_flow.py",
"accessCount": 2,
"lastAccessed": 1779386694151,
"type": "file"
},
{
"path": "examples\\01_connection\\async_connect.py",
"accessCount": 2,
"lastAccessed": 1779386740298,
"type": "file"
},
{
"path": "examples\\09_file_download\\report_file.py",
"accessCount": 2,
"lastAccessed": 1779386827437,
"type": "file"
},
{
"path": "examples\\10_offline\\block_data.py",
"accessCount": 2,
"lastAccessed": 1779386828134,
"type": "file"
},
{
"path": "examples/10_offline",
"accessCount": 2,
"lastAccessed": 1779386853771,
"type": "directory"
},
{
"path": "src\\easy_tdx\\models\\bar.py",
"accessCount": 2,
"lastAccessed": 1779389275981,
"type": "file"
},
{
"path": "src\\easy_tdx\\models\\stats.py",
"accessCount": 2,
"lastAccessed": 1779389276114,
"type": "file"
},
{
"path": "src\\easy_tdx\\models\\finance.py",
"accessCount": 2,
"lastAccessed": 1779389276205,
"type": "file"
},
{
"path": "examples",
"accessCount": 2,
"lastAccessed": 1779389346769,
"type": "directory"
},
{
"path": "src\\xmtdx\\__init__.py",
"accessCount": 1,
"lastAccessed": 1779337399287,
"type": "file"
},
{
"path": "src\\xmtdx\\client.py",
"accessCount": 1,
"lastAccessed": 1779337400354,
"type": "file"
},
{
"path": "src\\xmtdx\\models\\enums.py",
"accessCount": 1,
"lastAccessed": 1779337406304,
"type": "file"
},
{
"path": ".claude\\skills\\verify\\SKILL.md",
"accessCount": 1,
"lastAccessed": 1779337597439,
"type": "file"
},
{
"path": ".claude\\settings.json",
"accessCount": 1,
"lastAccessed": 1779337793420,
"type": "file"
},
{
"path": "C:\\new_jyplug\\connect.cfg",
"accessCount": 1,
"lastAccessed": 1779338269266,
"type": "file"
},
{
"path": "src\\xmtdx\\transport\\async_.py",
"accessCount": 1,
"lastAccessed": 1779338280552,
"type": "file"
},
{
"path": "src\\easy_tdx\\models\\__init__.py",
"accessCount": 1,
"lastAccessed": 1779384829108,
"type": "file"
},
{
"path": "src\\easy_tdx\\models\\security.py",
"accessCount": 1,
"lastAccessed": 1779384845552,
"type": "file"
},
{
"path": "src\\easy_tdx\\models\\quote.py",
"accessCount": 1,
"lastAccessed": 1779384845648,
"type": "file"
},
{
"path": "examples\\02_market_info\\security_count.py",
"accessCount": 1,
"lastAccessed": 1779386494822,
"type": "file"
},
{
"path": "examples\\01_connection\\ping_servers.py",
"accessCount": 1,
"lastAccessed": 1779386715403,
"type": "file"
},
{
"path": "examples\\01_connection\\connect_best_host.py",
"accessCount": 1,
"lastAccessed": 1779386715553,
"lastAccessed": 1781026550566,
"type": "file"
}
],
.omc/state/agent-replay-84b721c5-06e3-4191-ae03-5f4795cf33f3.jsonl
-5
View File
@@ -1,5 +0,0 @@
{"t":0,"agent":"aa8e8e8","agent_type":"unknown","event":"agent_stop","success":true}
{"t":0,"agent":"a64f368","agent_type":"unknown","event":"agent_stop","success":true}
{"t":0,"agent":"ad3555c","agent_type":"unknown","event":"agent_stop","success":true}
{"t":0,"agent":"aee6001","agent_type":"unknown","event":"agent_stop","success":true}
{"t":0,"agent":"a116063","agent_type":"unknown","event":"agent_stop","success":true}
.omc/state/hud-stdin-cache.json
+1 -1
View File
@@ -1 +1 @@
{"session_id":"84b721c5-06e3-4191-ae03-5f4795cf33f3","transcript_path":"C:\\Users\\毛利哥\\.claude\\projects\\D--python-easty-tdx\\84b721c5-06e3-4191-ae03-5f4795cf33f3.jsonl","cwd":"D:\\python\\easty_tdx","effort":{"level":"medium"},"session_name":"merge-datetime-fields-to-columns","model":{"id":"glm-5.1","display_name":"glm-5.1"},"workspace":{"current_dir":"D:\\python\\easty_tdx","project_dir":"D:\\python\\easty_tdx","added_dirs":[],"repo":{"host":"github.com","owner":"handsomejustin","name":"easy_tdx"}},"version":"2.1.145","output_style":{"name":"default"},"cost":{"total_cost_usd":19.846790000000016,"total_duration_ms":10309483,"total_api_duration_ms":4524776,"total_lines_added":739,"total_lines_removed":722},"context_window":{"total_input_tokens":88458,"total_output_tokens":37,"context_window_size":200000,"current_usage":{"input_tokens":3722,"output_tokens":37,"cache_creation_input_tokens":0,"cache_read_input_tokens":84736},"used_percentage":44,"remaining_percentage":56},"exceeds_200k_tokens":false,"fast_mode":false,"thinking":{"enabled":true}}
{"session_id":"86732a4c-5e66-4d81-afe8-a170fdd2102f","transcript_path":"C:\\Users\\毛利哥\\.claude\\projects\\D--python-easy-tdx\\86732a4c-5e66-4d81-afe8-a170fdd2102f.jsonl","cwd":"D:\\python\\easy_tdx","effort":{"level":"medium"},"session_name":"multi-factor-combo-backtest","model":{"id":"glm-5.1","display_name":"glm-5.1"},"workspace":{"current_dir":"D:\\python\\easy_tdx","project_dir":"D:\\python\\easy_tdx","added_dirs":[],"repo":{"host":"github.com","owner":"handsomejustin","name":"easy_tdx"}},"version":"2.1.165","output_style":{"name":"default"},"cost":{"total_cost_usd":5.7581050000000005,"total_duration_ms":77996980,"total_api_duration_ms":1457761,"total_lines_added":1393,"total_lines_removed":28},"context_window":{"total_input_tokens":134645,"total_output_tokens":42,"context_window_size":200000,"current_usage":{"input_tokens":245,"output_tokens":42,"cache_creation_input_tokens":0,"cache_read_input_tokens":134400},"used_percentage":67,"remaining_percentage":33},"exceeds_200k_tokens":false,"fast_mode":false,"thinking":{"enabled":true}}
.omc/state/last-tool-error.json
-7
View File
@@ -1,7 +0,0 @@
{
"tool_name": "Bash",
"tool_input_preview": "{\"command\":\"if exist \\\"D:\\\\python\\\\easty_tdx\\\\src\\\\xmtdx\\\" (echo \\\"src\\\\xmtdx EXISTS\\\" && dir \\\"D:\\\\python\\\\easty_tdx\\\\src\\\\xmtdx\\\") else (echo \\\"src\\\\xmtdx does NOT exist\\\")\"}",
"error": "Exit code 2\n/usr/bin/bash: eval: line 1: syntax error near unexpected token `('\n/usr/bin/bash: eval: line 1: `if exist \"D:\\python\\easty_tdx\\src\\xmtdx\" (echo \"src\\xmtdx EXISTS\" && dir \"D:\\python\\easty_tdx\\src\\xmtdx\") else (echo \"src\\xmtdx does NOT exist\")'",
"timestamp": "2026-05-21T20:17:49.142Z",
"retry_count": 1
}
.omc/state/sessions/84b721c5-06e3-4191-ae03-5f4795cf33f3/session-started.json
-6
View File
@@ -1,6 +0,0 @@
{
"session_id": "84b721c5-06e3-4191-ae03-5f4795cf33f3",
"started_at": "2026-05-21T19:45:15.057Z",
"cwd": "D:\\python\\easty_tdx",
"pid": 24604
}
.omc/state/subagent-tracking.json
-7
View File
@@ -1,7 +0,0 @@
{
"agents": [],
"total_spawned": 0,
"total_completed": 0,
"total_failed": 0,
"last_updated": "2026-05-21T20:14:12.044Z"
}
README.md
+55 -1
View File
@@ -300,6 +300,48 @@ python -X utf8 run_all_strategies.py SZ 300308 --count 2000 --cash 1000000 --adj
python -X utf8 run_all_strategies.py SZ 300308 --count 2000 --cash 1000000 --adjust QFQ --show
```
**多因子组合回测：**
自动遍历所有 2 因子 / 3 因子组合，找到最优搭配：
```bash
# 自动寻找最佳 2 因子和 3 因子组合（MAJORITY 模式）
python -X utf8 run_all_strategies.py SZ 300308 --combo 2 --combo 3 --combo-mode majority
# 也可用 AND / OR 模式
python -X utf8 run_all_strategies.py SZ 300308 --combo 2 --combo-mode and
```
CLI 指定策略文件组合：
```bash
easy-tdx backtest SZ 000001 \
--combo-strategies strategies/macd_cross.py,strategies/rsi_reversal.py,strategies/bollinger_breakout.py \
--combo-mode majority --table
```
Python API：
```python
from easy_tdx.backtest import CombinationRunner
runner = CombinationRunner(
strategy_classes=[MACDStrategy, RSIStrategy, BollingerStrategy],
df=df, cash=100000,
)
results = runner.screen(combo_sizes=(2, 3), mode="MAJORITY")
for r in results[:5]:
print(f"{r.name}: 收益={r.result.performance['total_return']:.2%}")
```
信号合并模式：
| 模式 | 买入条件 | 卖出条件 | 特点 |
|------|---------|---------|------|
| `AND` | 所有因子都看多 | 所有因子都看空 | 极保守，交易少但精确 |
| `MAJORITY` | 过半因子看多 | 过半因子看空 | 平衡，推荐默认 |
| `OR` | 任一因子看多 | 任一因子看空 | 激进，信号多噪声大 |
`--show` 会用 matplotlib 弹出一个双轴对比窗口：左轴蓝色线是归一化股价，右轴红色线是最佳策略的资金曲线，绿三角=买入、黄三角=卖出，标题显示股票名称和关键绩效指标。需要 `pip install matplotlib`。
输出示例（以 SZ 300308 为例）：
@@ -1087,7 +1129,7 @@ src/easy_tdx/
├── commands/ # 标准协议命令（无 IO）
├── codec/ # price / volume / datetime / frame / bitmap 编解码
├── chanlun/ # 缠论技术分析（K线合并/分型/笔/线段/中枢/买卖点/背驰）
├── backtest/ # 回测引擎（Strategy基类/向量化引擎/绩效分析）
├── backtest/ # 回测引擎（Strategy基类/向量化引擎/多因子组合/绩效分析）
├── models/ # 纯 dataclass，无业务逻辑
├── offline/ # 离线数据读写模块（读取 + 写入同步）
└── cli/ # easy-tdx CLI（click）
@@ -1116,6 +1158,18 @@ ruff format --check src/ tests/ # format check
## Changelog
### 1.9.0 (2026-06-10)
**多因子组合回测** — 新增组合回测引擎，支持 2-3 个因子信号叠加，自动遍历所有组合寻找最优搭配。
- 新增 `backtest/combo.py` 模块：`CombinationRunner`、`extract_factor_signals`、`combine_masks`、`FactorSignals`、`ComboResult`
- 信号合并模式：AND（全部同意）、OR（任一同意）、MAJORITY（过半同意）
- CLI 新增 `--combo-strategies` 和 `--combo-mode` 参数，支持指定策略文件组合回测
- `run_all_strategies.py` 新增 `--combo` 和 `--combo-mode` 选项，自动遍历 C(N,2)/C(N,3) 所有组合并排名
- 核心思路：预提取 N 个因子信号（只跑一次）→ 遍历组合合并遮罩（纯 numpy）→ 批量回测排名
- 新增 14 个单元测试（离线，无需网络）
- 修复 MyTT `MFI()` / `CR()` 指标分母为零时的 RuntimeWarning
### 1.8.2 (2026-06-09)
**策略扩充 + 可视化** — 新增 6 个策略（共 15 个）、`--show` 资金曲线图、茅台 demo 截图。
pyproject.toml
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project]
name = "easy-tdx"
version = "1.8.2"
version = "1.9.0"
description = "通达信 TCP 协议行情数据客户端，支持在线行情、离线数据读取与写入同步"
readme = "README.md"
requires-python = ">=3.10"
run_all_strategies.py
+140 -9
View File
@@ -59,6 +59,97 @@ def _map_trade_values(trades_df: Any, equity: Any, initial_cash: float) -> list[
return result_vals
def _run_combo_screen(
strategy_files: list[Path],
df: Any,
cash: float,
commission: float,
combo_sizes: tuple[int, ...],
combo_mode: str,
) -> None:
"""运行多因子组合回测并输出排名。"""
import importlib.util
from easy_tdx.backtest.combo import CombinationRunner
from easy_tdx.backtest.strategy import Strategy
# 加载所有策略类
strategy_classes: list[type[Strategy]] = []
for sf in strategy_files:
spec = importlib.util.spec_from_file_location("strategy_module", sf)
if spec is None or spec.loader is None:
continue
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
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:
strategy_classes.append(obj)
break
except TypeError:
pass
if len(strategy_classes) < 2:
click.echo("[!] 策略数量不足 2 个，跳过组合回测")
return
from math import comb
for size in combo_sizes:
total = comb(len(strategy_classes), size)
click.echo("\n" + "=" * 80)
click.echo(f"[*] {size}因子组合回测 (共{total}组, 模式={combo_mode})")
click.echo("=" * 80)
runner = CombinationRunner(
strategy_classes=strategy_classes,
df=df,
cash=cash,
commission=commission,
)
results = runner.screen(combo_sizes=(size,), mode=combo_mode.upper())
if not results:
click.echo(" 无有效交易组合（所有组合均为零交易）")
continue
# 表头
click.echo(
f"{'排名':>4} {'因子组合':<50} {'总收益率':>10} {'年化收益':>10} "
f"{'最大回撤':>10} {'夏普':>8} {'胜率':>8} {'交易':>6}"
)
click.echo("-" * 120)
for i, r in enumerate(results[:20], 1): # 只显示 top 20
medal = " *1*" if i == 1 else " *2*" if i == 2 else " *3*" if i == 3 else " "
perf = r.result.performance
click.echo(
f"{medal}{i:>2} {r.name:<50} "
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}"
)
if len(results) > 20:
click.echo(f" ... 共 {len(results)} 个有效组合，仅显示前 20")
# 最佳组合详细报告
best = results[0]
bp = best.result.performance
click.echo(f"\n[BEST {size}因子] {best.name}")
click.echo(f" 总收益率: {bp.get('total_return', 0):.2%}")
click.echo(f" 年化收益: {bp.get('annual_return', 0):.2%}")
click.echo(f" 最大回撤: {bp.get('max_drawdown', 0):.2%}")
click.echo(f" 夏普比率: {bp.get('sharpe', 0):.2f}")
click.echo(f" 胜率: {bp.get('win_rate', 0):.1%}")
def _show_best_chart(
df: Any,
result: Any,
@@ -95,8 +186,7 @@ def _show_best_chart(
eq_dates = equity["datetime"]
eq_values = equity["total"].values / initial_cash
ax2 = ax1.twinx()
ax2.plot(eq_dates, eq_values, color="crimson", linewidth=1.5,
label=f"策略: {strategy_name}")
ax2.plot(eq_dates, eq_values, color="crimson", linewidth=1.5, label=f"策略: {strategy_name}")
ax2.set_ylabel("资金曲线 (归一化)", color="crimson", fontsize=11)
ax2.tick_params(axis="y", labelcolor="crimson")
@@ -109,13 +199,23 @@ def _show_best_chart(
ax2.scatter(
buy_trades["datetime"].values,
_map_trade_values(buy_trades, equity, initial_cash),
marker="^", color="green", s=30, alpha=0.7, zorder=5, label="买入",
marker="^",
color="green",
s=30,
alpha=0.7,
zorder=5,
label="买入",
)
if not sell_trades.empty:
ax2.scatter(
sell_trades["datetime"].values,
_map_trade_values(sell_trades, equity, initial_cash),
marker="v", color="orange", s=30, alpha=0.7, zorder=5, label="卖出",
marker="v",
color="orange",
s=30,
alpha=0.7,
zorder=5,
label="卖出",
)
# 标题：股票代码 + 名称 + 策略绩效
@@ -150,6 +250,20 @@ def _show_best_chart(
@click.option("--commission", default=0.0003, type=float, help="佣金率")
@click.option("--adjust", default="QFQ", help="复权: NONE/QFQ/HFQ")
@click.option("--period", default="DAILY", help="K线周期")
@click.option(
"--combo",
"combo_sizes",
multiple=True,
type=int,
help="多因子组合回测（可多次指定，如 --combo 2 --combo 3）",
)
@click.option(
"--combo-mode",
"combo_mode",
default="MAJORITY",
type=click.Choice(["AND", "OR", "MAJORITY"], case_sensitive=False),
help="多因子信号合并模式（默认 MAJORITY）",
)
@click.option("--show", "show_chart", is_flag=True, help="显示最佳策略资金曲线 vs 股价对比图")
def run_all(
market: str,
@@ -159,6 +273,8 @@ def run_all(
commission: float,
adjust: str,
period: str,
combo_sizes: tuple[int, ...],
combo_mode: str,
show_chart: bool,
) -> None:
"""批量运行 strategies/ 目录下所有策略并比较结果。"""
@@ -253,7 +369,8 @@ def run_all(
perf = result.performance
click.echo(f" 完成 ({elapsed:.1f}s)")
results.append({
results.append(
{
"strategy": strategy_name,
"total_return": perf.get("total_return", 0),
"annual_return": perf.get("annual_return", 0),
@@ -265,15 +382,18 @@ def run_all(
"total_trades": perf.get("total_trades", 0),
"profit_factor": perf.get("profit_factor", 0),
"volatility": perf.get("volatility", 0),
})
}
)
backtest_results[strategy_name] = result
except Exception as e:
elapsed = time.perf_counter() - t0
click.echo(f" 错误 ({elapsed:.1f}s): {e}")
results.append({
results.append(
{
"strategy": strategy_name,
"error": str(e),
})
}
)
# 4. 输出排名
click.echo("\n" + "=" * 80)
@@ -400,7 +520,18 @@ def run_all(
for r in errored:
click.echo(f" {r['strategy']}: {r['error']}")
# 5. 展示最佳策略曲线图
# 5. 多因子组合回测
if combo_sizes:
_run_combo_screen(
strategy_files=strategy_files,
df=df,
cash=cash,
commission=commission,
combo_sizes=combo_sizes,
combo_mode=combo_mode,
)
# 6. 展示最佳策略曲线图
if show_chart and valid:
best_name = valid[0]["strategy"]
if best_name in backtest_results:
src/easy_tdx/MyTT.py
+220 -56
View File
@@ -22,288 +22,448 @@ import pandas as pd
# ------------------ 0级：核心工具函数 --------------------------------------------
def RD(N,D=3): return np.round(N,D) #四舍五入取3位小数
def RET(S,N=1): return np.array(S)[-N] #返回序列倒数第N个值,默认返回最后一个
def ABS(S): return np.abs(S) #返回N的绝对值
def LN(S): return np.log(S) #求底是e的自然对数,
def POW(S,N): return np.power(S,N) #求S的N次方
def SQRT(S): return np.sqrt(S) #求S的平方根
def SIN(S): return np.sin(S) #求S的正弦值（弧度)
def COS(S): return np.cos(S) #求S的余弦值（弧度)
def TAN(S): return np.tan(S) #求S的正切值（弧度)
def MAX(S1,S2): return np.maximum(S1,S2) #序列max
def MIN(S1,S2): return np.minimum(S1,S2) #序列min
def IF(S,A,B): return np.where(S,A,B) #序列布尔判断 return=A if S==True else B
def RD(N, D=3):
return np.round(N, D) # 四舍五入取3位小数
def RET(S, N=1):
return np.array(S)[-N] # 返回序列倒数第N个值,默认返回最后一个
def ABS(S):
return np.abs(S) # 返回N的绝对值
def LN(S):
return np.log(S) # 求底是e的自然对数,
def POW(S, N):
return np.power(S, N) # 求S的N次方
def SQRT(S):
return np.sqrt(S) # 求S的平方根
def SIN(S):
return np.sin(S) # 求S的正弦值（弧度)
def COS(S):
return np.cos(S) # 求S的余弦值（弧度)
def TAN(S):
return np.tan(S) # 求S的正切值（弧度)
def MAX(S1, S2):
return np.maximum(S1, S2) # 序列max
def MIN(S1, S2):
return np.minimum(S1, S2) # 序列min
def IF(S, A, B):
return np.where(S, A, B) # 序列布尔判断 return=A if S==True else B
def REF(S, N=1): # 对序列整体下移动N,返回序列(shift后会产生NAN)
return pd.Series(S).shift(N).values
def DIFF(S, N=1): # 前一个值减后一个值,前面会产生nan
return pd.Series(S).diff(N).values # np.diff(S)直接删除nan，会少一行
def STD(S, N): # 求序列的N日标准差，返回序列
return pd.Series(S).rolling(N).std(ddof=0).values
def SUM(S, N): # 对序列求N天累计和，返回序列 N=0对序列所有依次求和
return pd.Series(S).rolling(N).sum().values if N > 0 else pd.Series(S).cumsum().values
def CONST(S): # 返回序列S最后的值组成常量序列
return np.full(len(S), S[-1])
def HHV(S, N): # HHV(C, 5) 最近5天收盘最高价
return pd.Series(S).rolling(N).max().values
def LLV(S, N): # LLV(C, 5) 最近5天收盘最低价
return pd.Series(S).rolling(N).min().values
def HHVBARS(S, N): # 求N周期内S最高值到当前周期数, 返回序列
return pd.Series(S).rolling(N).apply(lambda x: np.argmax(x[::-1]), raw=True).values
def LLVBARS(S, N): # 求N周期内S最低值到当前周期数, 返回序列
return pd.Series(S).rolling(N).apply(lambda x: np.argmin(x[::-1]), raw=True).values
def MA(S, N): # 求序列的N日简单移动平均值，返回序列
return pd.Series(S).rolling(N).mean().values
def EMA(S, N): # 指数移动平均,为了精度 S>4*N EMA至少需要120周期 alpha=2/(span+1)
return pd.Series(S).ewm(span=N, adjust=False).mean().values
def SMA(S, N, M=1): # 中国式的SMA,至少需要120周期才精确 (雪球180周期) alpha=1/(1+com)
return pd.Series(S).ewm(alpha=M / N, adjust=False).mean().values # com=N-M/M
def WMA(S, N): # 通达信S序列的N日加权移动平均 Yn = (1*X1+2*X2+3*X3+...+n*Xn)/(1+2+3+...+Xn)
return pd.Series(S).rolling(N).apply(lambda x:x[::-1].cumsum().sum()*2/N/(N+1),raw=True).values
return (
pd.Series(S)
.rolling(N)
.apply(lambda x: x[::-1].cumsum().sum() * 2 / N / (N + 1), raw=True)
.values
)
def DMA(S, A): # 求S的动态移动平均，A作平滑因子,必须 0<A<1 (此为核心函数，非指标）
if isinstance(A,(int,float)): return pd.Series(S).ewm(alpha=A,adjust=False).mean().values
A=np.array(A); A[np.isnan(A)]=1.0; Y= np.zeros(len(S)); Y[0]=S[0]
for i in range(1,len(S)): Y[i]=A[i]*S[i]+(1-A[i])*Y[i-1] #A支持序列 by jqz1226
if isinstance(A, (int, float)):
return pd.Series(S).ewm(alpha=A, adjust=False).mean().values
A = np.array(A)
A[np.isnan(A)] = 1.0
Y = np.zeros(len(S))
Y[0] = S[0]
for i in range(1, len(S)):
Y[i] = A[i] * S[i] + (1 - A[i]) * Y[i - 1] # A支持序列 by jqz1226
return Y
def AVEDEV(S, N): # 平均绝对偏差 (序列与其平均值的绝对差的平均值)
return pd.Series(S).rolling(N).apply(lambda x: (np.abs(x - x.mean())).mean()).values
def SLOPE(S, N): # 返S序列N周期回线性回归斜率
return pd.Series(S).rolling(N).apply(lambda x: np.polyfit(range(N),x,deg=1)[0],raw=True).values
return (
pd.Series(S).rolling(N).apply(lambda x: np.polyfit(range(N), x, deg=1)[0], raw=True).values
)
def FORCAST(S, N): # 返回S序列N周期回线性回归后的预测值， jqz1226改进成序列出
return pd.Series(S).rolling(N).apply(lambda x:np.polyval(np.polyfit(range(N),x,deg=1),N-1),raw=True).values
return (
pd.Series(S)
.rolling(N)
.apply(lambda x: np.polyval(np.polyfit(range(N), x, deg=1), N - 1), raw=True)
.values
)
def LAST(S, A, B): # 从前A日到前B日一直满足S_BOOL条件, 要求A>B & A>0 & B>=0
return np.array(pd.Series(S).rolling(A+1).apply(lambda x:np.all(x[::-1][B:]),raw=True),dtype=bool)
return np.array(
pd.Series(S).rolling(A + 1).apply(lambda x: np.all(x[::-1][B:]), raw=True), dtype=bool
)
# ------------------ 1级：应用层函数(通过0级核心函数实现）使用方法请参考通达信--------------------------------
def COUNT(S, N): # COUNT(CLOSE>O, N): 最近N天满足S_BOO的天数 True的天数
return SUM(S, N)
def EVERY(S, N): # EVERY(CLOSE>O, 5) 最近N天是否都是True
return IF(SUM(S, N) == N, True, False)
def EXIST(S, N): # EXIST(CLOSE>3010, N=5) n日内是否存在一天大于3000点
return IF(SUM(S, N) > 0, True, False)
def FILTER(S, N): # FILTER函数，S满足条件后，将其后N周期内的数据置为0, FILTER(C==H,5)
for i in range(len(S)): S[i+1:i+1+N]=0 if S[i] else S[i+1:i+1+N]
for i in range(len(S)):
S[i + 1 : i + 1 + N] = 0 if S[i] else S[i + 1 : i + 1 + N]
return S # 例：FILTER(C==H,5) 涨停后，后5天不再发出信号
def BARSLAST(S): # 上一次条件成立到当前的周期, BARSLAST(C/REF(C,1)>=1.1) 上一次涨停到今天的天数
M = np.concatenate(([0], np.where(S, 1, 0)))
for i in range(1, len(M)): M[i]=0 if M[i] else M[i-1]+1
for i in range(1, len(M)):
M[i] = 0 if M[i] else M[i - 1] + 1
return M[1:]
def BARSLASTCOUNT(S): # 统计连续满足S条件的周期数 by jqz1226
rt = np.zeros(len(S) + 1) # BARSLASTCOUNT(CLOSE>OPEN)表示统计连续收阳的周期数
for i in range(len(S)): rt[i+1]=rt[i]+1 if S[i] else rt[i+1]
for i in range(len(S)):
rt[i + 1] = rt[i] + 1 if S[i] else rt[i + 1]
return rt[1:]
def BARSSINCEN(S, N): # N周期内第一次S条件成立到现在的周期数,N为常量 by jqz1226
return pd.Series(S).rolling(N).apply(lambda x:N-1-np.argmax(x) if np.argmax(x) or x[0] else 0,raw=True).fillna(0).values.astype(int)
return (
pd.Series(S)
.rolling(N)
.apply(lambda x: N - 1 - np.argmax(x) if np.argmax(x) or x[0] else 0, raw=True)
.fillna(0)
.values.astype(int)
)
def CROSS(S1, S2): # 判断向上金叉穿越 CROSS(MA(C,5),MA(C,10)) 判断向下死叉穿越 CROSS(MA(C,10),MA(C,5))
return np.concatenate(([False], np.logical_not((S1>S2)[:-1]) & (S1>S2)[1:])) # 不使用0级函数,移植方便 by jqz1226
def LONGCROSS(S1,S2,N): # 两条线维持一定周期后交叉,S1在N周期内都小于S2,本周期从S1下方向上穿过S2时返回1,否则返回0
return np.array(np.logical_and(LAST(S1<S2,N,1),(S1>S2)),dtype=bool) # N=1时等同于CROSS(S1, S2)
def CROSS(
S1, S2
): # 判断向上金叉穿越 CROSS(MA(C,5),MA(C,10)) 判断向下死叉穿越 CROSS(MA(C,10),MA(C,5))
return np.concatenate(
([False], np.logical_not((S1 > S2)[:-1]) & (S1 > S2)[1:])
) # 不使用0级函数,移植方便 by jqz1226
def LONGCROSS(
S1, S2, N
): # 两条线维持一定周期后交叉,S1在N周期内都小于S2,本周期从S1下方向上穿过S2时返回1,否则返回0
return np.array(
np.logical_and(LAST(S1 < S2, N, 1), (S1 > S2)), dtype=bool
) # N=1时等同于CROSS(S1, S2)
def VALUEWHEN(S, X): # 当S条件成立时,取X的当前值,否则取VALUEWHEN的上个成立时的X值 by jqz1226
return pd.Series(np.where(S, X, np.nan)).ffill().values
def BETWEEN(S, A, B): # S处于A和B之间时为真。 包括 A<S<B 或 A>S>B
return ((A < S) & (S < B)) | ((A > S) & (S > B))
def TOPRANGE(S): # TOPRANGE(HIGH)表示当前最高价是近多少周期内最高价的最大值 by jqz1226
rt = np.zeros(len(S))
for i in range(1,len(S)): rt[i] = np.argmin(np.flipud(S[:i]<S[i]))
return rt.astype('int')
for i in range(1, len(S)):
rt[i] = np.argmin(np.flipud(S[:i] < S[i]))
return rt.astype("int")
def LOWRANGE(S): # LOWRANGE(LOW)表示当前最低价是近多少周期内最低价的最小值 by jqz1226
rt = np.zeros(len(S))
for i in range(1,len(S)): rt[i] = np.argmin(np.flipud(S[:i]>S[i]))
return rt.astype('int')
for i in range(1, len(S)):
rt[i] = np.argmin(np.flipud(S[:i] > S[i]))
return rt.astype("int")
# ------------------ 2级：技术指标函数(全部通过0级，1级函数实现） ------------------------------
def MACD(CLOSE, SHORT=12, LONG=26, M=9): # EMA的关系，S取120日，和雪球小数点2位相同
DIF = EMA(CLOSE, SHORT) - EMA(CLOSE, LONG)
DEA = EMA(DIF,M); MACD=(DIF-DEA)*2
DEA = EMA(DIF, M)
MACD = (DIF - DEA) * 2
return RD(DIF), RD(DEA), RD(MACD)
def KDJ(CLOSE, HIGH, LOW, N=9, M1=3, M2=3): # KDJ指标
low_n = LLV(LOW, N)
high_n = HHV(HIGH, N)
high_low_diff = high_n - low_n
# 避免除零：当最高价等于最低价时，RSV 应该为 50（中性）
with np.errstate(divide='ignore', invalid='ignore'):
with np.errstate(divide="ignore", invalid="ignore"):
rsv = (CLOSE - low_n) / high_low_diff * 100
rsv = np.where(high_low_diff == 0, 50, rsv) # 除零时返回 50
K = EMA(rsv, (M1*2-1)); D = EMA(K,(M2*2-1)); J=K*3-D*2
K = EMA(rsv, (M1 * 2 - 1))
D = EMA(K, (M2 * 2 - 1))
J = K * 3 - D * 2
return K, D, J
def RSI(CLOSE, N=24): # RSI指标,和通达信小数点2位相同
DIF = CLOSE - REF(CLOSE, 1)
abs_dif_sma = SMA(ABS(DIF), N)
# 避免除零：当价格完全不变时，RSI 应该为 50（中性）
with np.errstate(divide='ignore', invalid='ignore'):
with np.errstate(divide="ignore", invalid="ignore"):
rsi_value = SMA(MAX(DIF, 0), N) / abs_dif_sma * 100
rsi_value = np.where(abs_dif_sma == 0, 50, rsi_value) # 除零时返回 50
return RD(rsi_value)
def WR(CLOSE, HIGH, LOW, N=10, N1=6): # W&R 威廉指标
high_n = HHV(HIGH, N)
low_n = LLV(LOW, N)
high_low_diff = high_n - low_n
with np.errstate(divide='ignore', invalid='ignore'):
with np.errstate(divide="ignore", invalid="ignore"):
wr = (high_n - CLOSE) / high_low_diff * 100
wr = np.where(high_low_diff == 0, 50, wr) # 除零时返回 50
high_n1 = HHV(HIGH, N1)
low_n1 = LLV(LOW, N1)
high_low_diff1 = high_n1 - low_n1
with np.errstate(divide='ignore', invalid='ignore'):
with np.errstate(divide="ignore", invalid="ignore"):
wr1 = (high_n1 - CLOSE) / high_low_diff1 * 100
wr1 = np.where(high_low_diff1 == 0, 50, wr1) # 除零时返回 50
return RD(wr), RD(wr1)
def BIAS(CLOSE, L1=6, L2=12, L3=24): # BIAS乖离率
BIAS1 = (CLOSE - MA(CLOSE, L1)) / MA(CLOSE, L1) * 100
BIAS2 = (CLOSE - MA(CLOSE, L2)) / MA(CLOSE, L2) * 100
BIAS3 = (CLOSE - MA(CLOSE, L3)) / MA(CLOSE, L3) * 100
return RD(BIAS1), RD(BIAS2), RD(BIAS3)
def BOLL(CLOSE, N=20, P=2): # BOLL指标，布林带
MID = MA(CLOSE, N)
UPPER = MID + STD(CLOSE, N) * P
LOWER = MID - STD(CLOSE, N) * P
return RD(UPPER), RD(MID), RD(LOWER)
def PSY(CLOSE, N=12, M=6):
PSY = COUNT(CLOSE > REF(CLOSE, 1), N) / N * 100
PSYMA = MA(PSY, M)
return RD(PSY), RD(PSYMA)
def CCI(CLOSE, HIGH, LOW, N=14):
TP = (HIGH + LOW + CLOSE) / 3
return (TP - MA(TP, N)) / (0.015 * AVEDEV(TP, N))
def ATR(CLOSE, HIGH, LOW, N=20): # 真实波动N日平均值
TR = MAX(MAX((HIGH - LOW), ABS(REF(CLOSE, 1) - HIGH)), ABS(REF(CLOSE, 1) - LOW))
return MA(TR, N)
def BBI(CLOSE, M1=3, M2=6, M3=12, M4=20): # BBI多空指标
return (MA(CLOSE, M1) + MA(CLOSE, M2) + MA(CLOSE, M3) + MA(CLOSE, M4)) / 4
def DMI(CLOSE, HIGH, LOW, M1=14, M2=6): # 动向指标：结果和同花顺，通达信完全一致
TR = SUM(MAX(MAX(HIGH - LOW, ABS(HIGH - REF(CLOSE, 1))), ABS(LOW - REF(CLOSE, 1))), M1)
HD = HIGH - REF(HIGH, 1); LD = REF(LOW, 1) - LOW
HD = HIGH - REF(HIGH, 1)
LD = REF(LOW, 1) - LOW
DMP = SUM(IF((HD > 0) & (HD > LD), HD, 0), M1)
DMM = SUM(IF((LD > 0) & (LD > HD), LD, 0), M1)
PDI = DMP * 100 / TR; MDI = DMM * 100 / TR
PDI = DMP * 100 / TR
MDI = DMM * 100 / TR
ADX = MA(ABS(MDI - PDI) / (PDI + MDI) * 100, M2)
ADXR = (ADX + REF(ADX, M2)) / 2
return PDI, MDI, ADX, ADXR
def TAQ(HIGH, LOW, N): # 唐安奇通道(海龟)交易指标，大道至简，能穿越牛熊
UP=HHV(HIGH,N); DOWN=LLV(LOW,N); MID=(UP+DOWN)/2
UP = HHV(HIGH, N)
DOWN = LLV(LOW, N)
MID = (UP + DOWN) / 2
return UP, MID, DOWN
def KTN(CLOSE, HIGH, LOW, N=20, M=10): # 肯特纳交易通道, N选20日，ATR选10日
MID = EMA((HIGH + LOW + CLOSE) / 3, N)
ATRN = ATR(CLOSE, HIGH, LOW, M)
UPPER=MID+2*ATRN; LOWER=MID-2*ATRN
UPPER = MID + 2 * ATRN
LOWER = MID - 2 * ATRN
return UPPER, MID, LOWER
def TRIX(CLOSE, M1=12, M2=20): # 三重指数平滑平均线
TR = EMA(EMA(EMA(CLOSE, M1), M1), M1)
TRIX = (TR - REF(TR, 1)) / REF(TR, 1) * 100
TRMA = MA(TRIX, M2)
return TRIX, TRMA
def VR(CLOSE, VOL, M1=26): # VR容量比率
LC = REF(CLOSE, 1)
return SUM(IF(CLOSE > LC, VOL, 0), M1) / SUM(IF(CLOSE <= LC, VOL, 0), M1) * 100
def CR(CLOSE, HIGH, LOW, N=20): # CR价格动量指标
MID = REF(HIGH + LOW + CLOSE, 1) / 3
return SUM(MAX(0,HIGH-MID),N)/SUM(MAX(0,MID-LOW),N)*100
num = SUM(MAX(0, HIGH - MID), N)
den = SUM(MAX(0, MID - LOW), N)
with np.errstate(divide="ignore", invalid="ignore"):
return np.where(den > 0, num / den * 100, 100)
def EMV(HIGH, LOW, VOL, N=14, M=9): # 简易波动指标
VOLUME=MA(VOL,N)/VOL; MID=100*(HIGH+LOW-REF(HIGH+LOW,1))/(HIGH+LOW)
EMV=MA(MID*VOLUME*(HIGH-LOW)/MA(HIGH-LOW,N),N); MAEMV=MA(EMV,M)
VOLUME = MA(VOL, N) / VOL
MID = 100 * (HIGH + LOW - REF(HIGH + LOW, 1)) / (HIGH + LOW)
EMV = MA(MID * VOLUME * (HIGH - LOW) / MA(HIGH - LOW, N), N)
MAEMV = MA(EMV, M)
return EMV, MAEMV
def DPO(CLOSE, M1=20, M2=10, M3=6): # 区间震荡线
DPO = CLOSE - REF(MA(CLOSE, M1), M2); MADPO = MA(DPO, M3)
DPO = CLOSE - REF(MA(CLOSE, M1), M2)
MADPO = MA(DPO, M3)
return DPO, MADPO
def BRAR(OPEN, CLOSE, HIGH, LOW, M1=26): # BRAR-ARBR 情绪指标
AR = SUM(HIGH - OPEN, M1) / SUM(OPEN - LOW, M1) * 100
BR = SUM(MAX(0, HIGH - REF(CLOSE, 1)), M1) / SUM(MAX(0, REF(CLOSE, 1) - LOW), M1) * 100
return AR, BR
def DFMA(CLOSE, N1=10, N2=50, M=10): # 平行线差指标
DIF=MA(CLOSE,N1)-MA(CLOSE,N2); DIFMA=MA(DIF,M) #通达信指标叫DMA 同花顺叫新DMA
DIF = MA(CLOSE, N1) - MA(CLOSE, N2)
DIFMA = MA(DIF, M) # 通达信指标叫DMA 同花顺叫新DMA
return DIF, DIFMA
def MTM(CLOSE, N=12, M=6): # 动量指标
MTM=CLOSE-REF(CLOSE,N); MTMMA=MA(MTM,M)
MTM = CLOSE - REF(CLOSE, N)
MTMMA = MA(MTM, M)
return MTM, MTMMA
def MASS(HIGH, LOW, N1=9, N2=25, M=6): # 梅斯线
MASS = SUM(MA(HIGH - LOW, N1) / MA(MA(HIGH - LOW, N1), N1), N2)
MA_MASS = MA(MASS, M)
return MASS, MA_MASS
def ROC(CLOSE, N=12, M=6): # 变动率指标
ROC=100*(CLOSE-REF(CLOSE,N))/REF(CLOSE,N); MAROC=MA(ROC,M)
ROC = 100 * (CLOSE - REF(CLOSE, N)) / REF(CLOSE, N)
MAROC = MA(ROC, M)
return ROC, MAROC
def EXPMA(CLOSE, N1=12, N2=50): # EMA指数平均数指标
return EMA(CLOSE, N1), EMA(CLOSE, N2)
def OBV(CLOSE, VOL): # 能量潮指标
return SUM(IF(CLOSE > REF(CLOSE, 1), VOL, IF(CLOSE < REF(CLOSE, 1), -VOL, 0)), 0) / 10000
def MFI(CLOSE, HIGH, LOW, VOL, N=14): # MFI指标是成交量的RSI指标
TYP = (HIGH + LOW + CLOSE) / 3
V1=SUM(IF(TYP>REF(TYP,1),TYP*VOL,0),N)/SUM(IF(TYP<REF(TYP,1),TYP*VOL,0),N)
pos_mf = SUM(IF(TYP > REF(TYP, 1), TYP * VOL, 0), N)
neg_mf = SUM(IF(TYP < REF(TYP, 1), TYP * VOL, 0), N)
with np.errstate(divide="ignore", invalid="ignore"):
V1 = np.where(neg_mf > 0, pos_mf / neg_mf, np.where(pos_mf > 0, np.inf, 0))
return 100 - (100 / (1 + V1))
def ASI(OPEN, CLOSE, HIGH, LOW, M1=26, M2=10): # 振动升降指标
LC=REF(CLOSE,1); AA=ABS(HIGH-LC); BB=ABS(LOW-LC)
CC=ABS(HIGH-REF(LOW,1)); DD=ABS(LC-REF(OPEN,1))
R=IF( (AA>BB) & (AA>CC),AA+BB/2+DD/4,IF( (BB>CC) & (BB>AA),BB+AA/2+DD/4,CC+DD/4))
X=(CLOSE-LC+(CLOSE-OPEN)/2+LC-REF(OPEN,1))
SI=16*X/R*MAX(AA,BB); ASI=SUM(SI,M1); ASIT=MA(ASI,M2)
LC = REF(CLOSE, 1)
AA = ABS(HIGH - LC)
BB = ABS(LOW - LC)
CC = ABS(HIGH - REF(LOW, 1))
DD = ABS(LC - REF(OPEN, 1))
R = IF(
(AA > BB) & (AA > CC),
AA + BB / 2 + DD / 4,
IF((BB > CC) & (BB > AA), BB + AA / 2 + DD / 4, CC + DD / 4),
)
X = CLOSE - LC + (CLOSE - OPEN) / 2 + LC - REF(OPEN, 1)
SI = 16 * X / R * MAX(AA, BB)
ASI = SUM(SI, M1)
ASIT = MA(ASI, M2)
return ASI, ASIT
def XSII(CLOSE, HIGH, LOW, N=102, M=7): # 薛斯通道II
AA = MA((2 * CLOSE + HIGH + LOW) / 4, 5) # 最新版DMA才支持 2021-12-4
TD1 = AA*N/100; TD2 = AA*(200-N) / 100
TD1 = AA * N / 100
TD2 = AA * (200 - N) / 100
CC = ABS((2 * CLOSE + HIGH + LOW) / 4 - MA(CLOSE, 20)) / MA(CLOSE, 20)
DD = DMA(CLOSE,CC); TD3=(1+M/100)*DD; TD4=(1-M/100)*DD
DD = DMA(CLOSE, CC)
TD3 = (1 + M / 100) * DD
TD4 = (1 - M / 100) * DD
return TD1, TD2, TD3, TD4
@@ -315,20 +475,24 @@ def ZHUOYAO(CLOSE, N1=120, N2=60, N3=20, M=10): #捉妖大师指标：中长
TREND = EMA(MID, M) # 趋势 EXPMA(中线,10)
return RD(LONG), RD(MID), RD(SHORT), RD(TREND)
def BIAS_SIGNAL(CLOSE, P=10, M=30): # 乖离率信号指标：M日乖离 + 短/长信号线趋势判断
X = (CLOSE - MA(CLOSE, M)) / MA(CLOSE, M) * 100 # M日乖离率
S_SMA = MA(X, P) # 短周期信号线 MA(X,P)
X_LMA = MA(X, M) # 长周期信号线 MA(X,M)
return RD(X), RD(S_SMA), RD(X_LMA)
def FK(CLOSE): # FK趋势指标：快线EMA(2)与斜率外推慢线EMA(42)比较
fast = EMA(CLOSE, 2)
slow = EMA(SLOPE(CLOSE, 21) * 20 + CLOSE, 42)
return fast > slow
def OUTPERFORM_20D(CLOSE, INDEX_CLOSE): # 20日相对强度：个股涨幅跑赢大盘返回1，否则返回0
stock_ret = (CLOSE - REF(CLOSE, 20)) / REF(CLOSE, 20)
index_ret = (INDEX_CLOSE - REF(INDEX_CLOSE, 20)) / REF(INDEX_CLOSE, 20)
return IF(stock_ret > index_ret, 1, 0)
# 望大家能提交更多指标和函数 https://github.com/mpquant/MyTT
src/easy_tdx/backtest/__init__.py
+4
View File
@@ -20,6 +20,7 @@
print(result.performance)
"""
from easy_tdx.backtest.combo import CombinationRunner, ComboResult, FactorSignals # noqa: F401
from easy_tdx.backtest.engine import BacktestEngine # noqa: F401
from easy_tdx.backtest.strategy import Strategy, StrategyDataProxy, crossover # noqa: F401
from easy_tdx.backtest.types import BacktestResult, Position, Signal, Trade # noqa: F401
@@ -27,6 +28,9 @@ from easy_tdx.backtest.types import BacktestResult, Position, Signal, Trade # n
__all__ = [
"BacktestEngine",
"BacktestResult",
"CombinationRunner",
"ComboResult",
"FactorSignals",
"Strategy",
"StrategyDataProxy",
"Signal",
src/easy_tdx/backtest/cli.py
+70 -10
View File
@@ -14,6 +14,19 @@ import click
@click.argument("code")
@click.option("--strategy", "strategy_str", default=None, help="DSL 策略表达式 (P1)")
@click.option("--strategy-file", "strategy_file", default=None, help="Python 策略文件路径")
@click.option(
"--combo-strategies",
"combo_strategies",
default=None,
help="多因子组合：逗号分隔的策略文件路径（如 strats/a.py,strats/b.py,strats/c.py）",
)
@click.option(
"--combo-mode",
"combo_mode",
default="MAJORITY",
type=click.Choice(["AND", "OR", "MAJORITY"], case_sensitive=False),
help="多因子信号合并模式（默认 MAJORITY）",
)
@click.option("--cash", default=100000.0, type=float, help="初始资金")
@click.option("--commission", default=0.0003, type=float, help="佣金率")
@click.option(
@@ -33,6 +46,8 @@ def backtest(
code: str,
strategy_str: str | None,
strategy_file: str | None,
combo_strategies: str | None,
combo_mode: str,
cash: float,
commission: float,
execution: str,
@@ -52,16 +67,27 @@ def backtest(
easy-tdx backtest SH 600519 --strategy-file ma_cross.py --table
easy-tdx backtest SZ 000001 --strategy-file my_strategy.py --indicators MACD,KDJ
easy-tdx backtest SZ 000001 \
--combo-strategies strategies/macd_cross.py,strategies/rsi_reversal.py \
--combo-mode MAJORITY --table
"""
from ..backtest.engine import BacktestEngine
from ..cli.conn import get_mac_client
from ..cli.parsers import parse_adjust, parse_market, parse_period
from ..indicator import compute_indicators
# 1. 加载策略
# 1. 加载策略（单策略 or 多因子组合）
is_combo = combo_strategies is not None
strategy = None
if is_combo:
strategy = _load_combo_strategies(combo_strategies)
else:
strategy = _load_strategy(strategy_str, strategy_file)
if strategy is None:
click.echo("错误: 必须指定 --strategy-file 或 --strategy", err=True)
click.echo("错误: 必须指定 --strategy-file / --combo-strategies / --strategy", err=True)
raise SystemExit(1)
# 2. 获取数据
@@ -82,6 +108,22 @@ def backtest(
df = compute_indicators(df, indicator_list)
# 4. 创建引擎并运行
if is_combo:
from ..backtest.combo import CombinationRunner
assert strategy is not None # for type checker
runner = CombinationRunner(
strategy_classes=strategy,
df=df,
cash=cash,
commission=commission,
execution=execution,
)
result = runner.run_combination(
indices=list(range(len(strategy))),
mode=combo_mode.upper(),
)
else:
engine = BacktestEngine(
strategy=strategy,
cash=cash,
@@ -100,9 +142,7 @@ def backtest(
click.echo(result.to_json())
def _load_strategy(
strategy_str: str | None, strategy_file: str | None
) -> type | None:
def _load_strategy(strategy_str: str | None, strategy_file: str | None) -> type | None:
"""加载策略类。
优先从 Python 文件加载，其次从 DSL 表达式加载（未实现）。
@@ -154,11 +194,7 @@ def _load_strategy_from_file(path: str) -> type:
for name in dir(module):
obj = getattr(module, name)
try:
if (
isinstance(obj, type)
and issubclass(obj, Strategy)
and obj is not Strategy
):
if isinstance(obj, type) and issubclass(obj, Strategy) and obj is not Strategy:
strategy_classes.append(obj)
except TypeError:
pass
@@ -173,6 +209,30 @@ def _load_strategy_from_file(path: str) -> type:
return strategy_classes[0]
def _load_combo_strategies(combo_strategies: str) -> list[type]:
"""从逗号分隔的路径列表加载多个策略类。
Args:
combo_strategies: 逗号分隔的策略文件路径
Returns:
Strategy 子类列表
"""
paths = [p.strip() for p in combo_strategies.split(",") if p.strip()]
if len(paths) < 2:
click.echo("错误: --combo-strategies 至少需要 2 个策略文件", err=True)
raise SystemExit(1)
classes: list[type] = []
for p in paths:
cls = _load_strategy_from_file(p)
classes.append(cls)
names = [c.__name__ for c in classes]
click.echo(f"[*] 多因子组合 ({len(classes)} 因子): {' + '.join(names)}")
return classes
def _print_table(result: Any) -> None:
"""以表格形式输出回测结果。"""
perf = result.performance
src/easy_tdx/backtest/combo.py
+447
View File
@@ -0,0 +1,447 @@
"""多因子组合回测引擎。
核心能力：
- extract_factor_signals: 从策略提取买入/卖出信号遮罩
- combine_masks: 合并多个因子的信号（AND / OR / MAJORITY）
- CombinationRunner: 批量遍历因子组合，自动寻找最优搭配
用法::
from easy_tdx.backtest.combo import CombinationRunner
runner = CombinationRunner(
strategy_classes=[MACDStrategy, RSIStrategy, BollingerStrategy],
df=df,
cash=100000.0,
)
# 遍历所有 2/3 因子组合
results = runner.screen(combo_sizes=(2, 3), mode="MAJORITY")
for r in results[:5]:
print(f"{r.name}: 收益={r.result.performance['total_return']:.2%}")
"""
from __future__ import annotations
import itertools
from dataclasses import dataclass
from typing import Any
import numpy as np
import numpy.typing as npt
import pandas as pd
from easy_tdx.backtest.engine import BacktestEngine
from easy_tdx.backtest.strategy import Strategy
from easy_tdx.backtest.types import BacktestResult
NDArray = np.ndarray
BoolArray = npt.NDArray[np.bool_]
# ── 数据结构 ────────────────────────────────────────────────────────────────
@dataclass
class FactorSignals:
"""单个因子的信号遮罩。
Attributes:
name: 因子名称（通常为策略类名）
buy_mask: 每根 bar 是否产生买入信号
sell_mask: 每根 bar 是否产生卖出信号
"""
name: str
buy_mask: BoolArray
sell_mask: BoolArray
@dataclass
class ComboResult:
"""因子组合的回测结果。
Attributes:
name: 组合名称（如 "MACDStrategy + RSIStrategy"）
indices: 因子在原始列表中的索引
size: 因子数量
result: 回测结果
"""
name: str
indices: tuple[int, ...]
size: int
result: BacktestResult
# ── 信号提取 ────────────────────────────────────────────────────────────────
def extract_factor_signals(
strategy_cls: type[Strategy],
df: pd.DataFrame,
cash: float = 100_000.0,
commission: float = 0.0003,
) -> FactorSignals:
"""从策略类提取买入/卖出信号遮罩。
运行策略的 bar-by-bar 信号生成，捕获每根 bar 的买卖意图。
复现 BacktestEngine._generate_signals 的仓位跟踪逻辑，
确保信号与实际运行一致。
Args:
strategy_cls: Strategy 子类
df: K线 DataFrame
cash: 初始资金（影响全仓计算）
commission: 佣金率
Returns:
FactorSignals 包含 buy_mask 和 sell_mask
"""
strat = strategy_cls()
strat._bind_data(df)
strat._cash = cash
strat._position_size = 0.0
strat._call_init()
n = len(df)
buy_mask = np.zeros(n, dtype=bool)
sell_mask = np.zeros(n, dtype=bool)
close_arr = df["close"].to_numpy()
for i in range(n):
strat._set_bar_index(i)
strat._call_next()
signals = strat._clear_signals()
for sig in signals:
if sig.direction == "BUY":
buy_mask[i] = True
else:
sell_mask[i] = True
# 跟踪仓位状态（与 BacktestEngine._update_strategy_position 一致）
_update_position(strat, signals, close_arr[i], commission)
return FactorSignals(
name=strategy_cls.__name__,
buy_mask=buy_mask,
sell_mask=sell_mask,
)
def _update_position(
strat: Strategy,
signals: list[Any],
est_price: float,
commission: float,
) -> None:
"""更新策略内部仓位状态。
复现 BacktestEngine._update_strategy_position 的逻辑，
使因子信号提取与实际回测行为一致。
Args:
strat: 策略实例
signals: 当前 bar 的信号列表
est_price: 估算价格（收盘价）
commission: 佣金率
"""
for sig in signals:
price = sig.price or est_price
if sig.direction == "BUY":
if sig.size == 0:
shares = int(strat._cash / (price * (1 + commission)) / 100) * 100
if shares > 0:
strat._position_size += shares
strat._cash -= shares * price
else:
strat._position_size += sig.size
strat._cash -= sig.size * price
elif sig.direction == "SELL":
if sig.size == 0:
strat._cash += strat._position_size * price
strat._position_size = 0.0
else:
strat._cash += sig.size * price
strat._position_size = max(0.0, strat._position_size - sig.size)
# ── 信号合并 ────────────────────────────────────────────────────────────────
def combine_masks(
signals_list: list[FactorSignals],
mode: str = "MAJORITY",
) -> tuple[BoolArray, BoolArray]:
"""合并多个因子的信号遮罩。
Args:
signals_list: 因子信号列表
mode: 合并模式
- "AND": 所有因子都同意才触发
- "OR": 任一因子同意即触发
- "MAJORITY": 过半因子同意才触发
Returns:
(combined_buy_mask, combined_sell_mask)
Raises:
ValueError: 不支持的合并模式
"""
if not signals_list:
raise ValueError("至少需要 1 个因子信号")
buy_arrays = [s.buy_mask for s in signals_list]
sell_arrays = [s.sell_mask for s in signals_list]
buy_stack = np.stack(buy_arrays) # shape: (n_factors, n_bars)
sell_stack = np.stack(sell_arrays)
if mode == "AND":
return np.all(buy_stack, axis=0), np.all(sell_stack, axis=0)
elif mode == "OR":
return np.any(buy_stack, axis=0), np.any(sell_stack, axis=0)
elif mode == "MAJORITY":
n_factors = len(signals_list)
threshold = n_factors / 2
return (
np.sum(buy_stack, axis=0) > threshold,
np.sum(sell_stack, axis=0) > threshold,
)
else:
raise ValueError(f"不支持的合并模式: {mode!r}（可选: AND, OR, MAJORITY）")
# ── 组合策略包装 ─────────────────────────────────────────────────────────────
class _ComboStrategy(Strategy):
"""将合并后的信号遮罩包装为 Strategy 子类。
与 dsl_strategy 思路一致，但增加了仓位检查，
避免重复买入和空仓卖出。
"""
_buy_mask: BoolArray
_sell_mask: BoolArray
def init(self) -> None:
pass
def next(self) -> None:
idx = self._bar_index
buy = self._buy_mask
sell = self._sell_mask
if idx < len(buy) and buy[idx] and self.position["size"] == 0:
self.buy(size=0)
elif idx < len(sell) and sell[idx] and self.position["size"] > 0:
self.sell(size=0)
def _make_combo_strategy(
buy_mask: BoolArray,
sell_mask: BoolArray,
) -> type[_ComboStrategy]:
"""将信号遮罩包装为 _ComboStrategy 子类。
Args:
buy_mask: 合并后的买入遮罩
sell_mask: 合并后的卖出遮罩
Returns:
_ComboStrategy 子类
"""
class WrappedComboStrategy(_ComboStrategy):
_buy_mask = buy_mask
_sell_mask = sell_mask
return WrappedComboStrategy
# ── 组合回测运行器 ──────────────────────────────────────────────────────────
class CombinationRunner:
"""多因子组合回测运行器。
用法::
runner = CombinationRunner(
strategy_classes=[MACDStrategy, RSIStrategy, BollingerStrategy],
df=df,
cash=100000.0,
)
# 遍历所有 2 因子组合
results = runner.screen(combo_sizes=(2,), mode="MAJORITY")
"""
def __init__(
self,
strategy_classes: list[type[Strategy]],
df: pd.DataFrame,
cash: float = 100_000.0,
commission: float = 0.0003,
min_commission: float = 5.0,
stamp_tax: float = 0.001,
slippage: float = 0.0,
execution: str = "next_open",
position_mode: str = "full",
reject_policy: str = "reduce",
) -> None:
"""初始化运行器。
Args:
strategy_classes: 参与组合的策略类列表
df: K线数据（所有策略共享）
cash: 初始资金
commission: 佣金率
min_commission: 最低佣金
stamp_tax: 印花税率
slippage: 滑点
execution: 成交价规则
position_mode: 持仓模式
reject_policy: 拒单策略
"""
self._strategy_classes = strategy_classes
self._df = df
self._cash = cash
self._commission = commission
self._min_commission = min_commission
self._stamp_tax = stamp_tax
self._slippage = slippage
self._execution = execution
self._position_mode = position_mode
self._reject_policy = reject_policy
# 缓存：策略类 → FactorSignals（只提取一次）
self._signal_cache: dict[str, FactorSignals] = {}
def _get_factor_signals(self, idx: int) -> FactorSignals:
"""获取指定策略的因子信号（带缓存）。
Args:
idx: 策略在列表中的索引
Returns:
FactorSignals
"""
cls = self._strategy_classes[idx]
key = cls.__name__
if key not in self._signal_cache:
self._signal_cache[key] = extract_factor_signals(
cls, self._df, cash=self._cash, commission=self._commission
)
return self._signal_cache[key]
def _make_engine(self, strategy: type[Strategy]) -> BacktestEngine:
"""创建配置一致的回测引擎。
Args:
strategy: 策略类
Returns:
BacktestEngine 实例
"""
return BacktestEngine(
strategy=strategy,
cash=self._cash,
commission=self._commission,
min_commission=self._min_commission,
stamp_tax=self._stamp_tax,
slippage=self._slippage,
execution=self._execution,
position_mode=self._position_mode,
reject_policy=self._reject_policy,
)
def run_combination(
self,
indices: list[int] | tuple[int, ...],
mode: str = "MAJORITY",
) -> BacktestResult:
"""运行指定因子组合的回测。
Args:
indices: 要组合的因子索引（在 strategy_classes 中的位置）
mode: 信号合并模式（AND / OR / MAJORITY）
Returns:
BacktestResult
"""
# 1. 提取因子信号
signals = [self._get_factor_signals(i) for i in indices]
# 2. 合并信号
buy_mask, sell_mask = combine_masks(signals, mode=mode)
# 3. 包装为策略并运行回测
combo_cls = _make_combo_strategy(buy_mask, sell_mask)
engine = self._make_engine(combo_cls)
return engine.run(self._df)
def screen(
self,
combo_sizes: tuple[int, ...] = (2, 3),
mode: str = "MAJORITY",
filter_zero_trades: bool = True,
top_n: int = 0,
) -> list[ComboResult]:
"""遍历所有因子组合，批量回测并排名。
Args:
combo_sizes: 要尝试的组合大小（如 (2, 3) 表示 2 因子和 3 因子组合）
mode: 信号合并模式（AND / OR / MAJORITY）
filter_zero_trades: 是否过滤零交易组合
top_n: 只返回前 N 名（0 = 全部返回）
Returns:
按 total_return 降序排列的 ComboResult 列表
"""
n_factors = len(self._strategy_classes)
results: list[ComboResult] = []
for size in combo_sizes:
if size > n_factors:
continue
for combo in itertools.combinations(range(n_factors), size):
signals = [self._get_factor_signals(i) for i in combo]
# 合并信号
buy_mask, sell_mask = combine_masks(signals, mode=mode)
# 包装为策略并运行
combo_cls = _make_combo_strategy(buy_mask, sell_mask)
engine = self._make_engine(combo_cls)
result = engine.run(self._df)
name = " + ".join(s.name for s in signals)
results.append(
ComboResult(
name=name,
indices=combo,
size=size,
result=result,
)
)
# 过滤零交易
if filter_zero_trades:
results = [r for r in results if r.result.performance["total_trades"] > 0]
# 按总收益率降序排列
results.sort(key=lambda r: r.result.performance["total_return"], reverse=True)
# 截取 top_n
if top_n > 0:
results = results[:top_n]
return results
tests/unit/test_backtest_combo.py
+396
View File
@@ -0,0 +1,396 @@
"""测试多因子组合回测模块。"""
from __future__ import annotations
import numpy as np
import pandas as pd
import pytest
from easy_tdx import MyTT
from easy_tdx.backtest.strategy import Strategy
# ── 辅助函数 ─────────────────────────────────────────────────────────────────────
def _make_df(n: int = 100, seed: int = 42) -> pd.DataFrame:
"""构造合成 OHLCV DataFrame。"""
rng = np.random.default_rng(seed)
close = 100.0 + np.cumsum(rng.normal(0, 1, n))
high = close + rng.uniform(0, 1, n)
low = close - rng.uniform(0, 1, n)
open_ = low + rng.uniform(0, high - low, n)
volume = rng.integers(1_000_000, 10_000_000, n)
dates = pd.date_range("2024-01-01", periods=n, freq="D")
return pd.DataFrame(
{
"datetime": dates,
"open": open_,
"high": high,
"low": low,
"close": close,
"volume": volume,
}
)
# ── 测试用策略 ─────────────────────────────────────────────────────────────────
class BuyOnBar3Strategy(Strategy):
"""在第 3 根 bar 买入，第 8 根 bar 卖出（确定性策略，方便断言）。"""
def init(self) -> None:
pass
def next(self) -> None:
if self._bar_index == 3 and self.position["size"] == 0:
self.buy(size=0)
elif self._bar_index == 8 and self.position["size"] > 0:
self.sell(size=0)
class BuyOnBar5Strategy(Strategy):
"""在第 5 根 bar 买入，第 10 根 bar 卖出。"""
def init(self) -> None:
pass
def next(self) -> None:
if self._bar_index == 5 and self.position["size"] == 0:
self.buy(size=0)
elif self._bar_index == 10 and self.position["size"] > 0:
self.sell(size=0)
class BuyOnBar7Strategy(Strategy):
"""在第 7 根 bar 买入，第 12 根 bar 卖出。"""
def init(self) -> None:
pass
def next(self) -> None:
if self._bar_index == 7 and self.position["size"] == 0:
self.buy(size=0)
elif self._bar_index == 12 and self.position["size"] > 0:
self.sell(size=0)
class NeverTradeStrategy(Strategy):
"""从不交易的策略。"""
def init(self) -> None:
pass
def next(self) -> None:
pass
# ── extract_factor_signals 测试 ──────────────────────────────────────────────
class TestExtractFactorSignals:
"""测试因子信号提取。"""
def test_basic_buy_sell_signals(self) -> None:
"""BuyOnBar3Strategy 应在第 3 根 bar 买入、第 8 根 bar 卖出。"""
from easy_tdx.backtest.combo import extract_factor_signals
df = _make_df()
fs = extract_factor_signals(BuyOnBar3Strategy, df)
assert fs.name == "BuyOnBar3Strategy"
assert len(fs.buy_mask) == len(df)
assert len(fs.sell_mask) == len(df)
# 第 3 根 bar 买入
assert fs.buy_mask[3] is np.True_
# 其他 bar 不买入
buy_count = int(np.sum(fs.buy_mask))
assert buy_count == 1
# 第 8 根 bar 卖出
assert fs.sell_mask[8] is np.True_
sell_count = int(np.sum(fs.sell_mask))
assert sell_count == 1
def test_never_trade_strategy(self) -> None:
"""从不交易的策略应产生全空遮罩。"""
from easy_tdx.backtest.combo import extract_factor_signals
df = _make_df()
fs = extract_factor_signals(NeverTradeStrategy, df)
assert int(np.sum(fs.buy_mask)) == 0
assert int(np.sum(fs.sell_mask)) == 0
def test_real_ma_strategy(self) -> None:
"""真实 MA 策略的信号提取应有交易信号。"""
from easy_tdx.backtest.combo import extract_factor_signals
# 用较长数据确保有交叉
df = _make_df(n=200, seed=123)
class MAStrategy(Strategy):
def init(self) -> None:
self.ma5 = self.I(MyTT.MA, self.data.close, 5)
self.ma20 = self.I(MyTT.MA, self.data.close, 20)
def next(self) -> None:
if self._bar_index < 1:
return
prev_ma5 = self.ma5[self._bar_index - 1]
prev_ma20 = self.ma20[self._bar_index - 1]
curr_ma5 = self.ma5[self._bar_index]
curr_ma20 = self.ma20[self._bar_index]
if prev_ma5 <= prev_ma20 and curr_ma5 > curr_ma20:
self.buy(size=0)
elif prev_ma5 >= prev_ma20 and curr_ma5 < curr_ma20:
if self.position["size"] > 0:
self.sell(size=0)
fs = extract_factor_signals(MAStrategy, df)
# 200 bar 数据应该至少有一次买卖
assert int(np.sum(fs.buy_mask)) >= 1
assert int(np.sum(fs.sell_mask)) >= 0 # 可能没有卖（最后还在持仓）
# ── combine_masks 测试 ──────────────────────────────────────────────────────
class TestCombineMasks:
"""测试信号遮罩合并。"""
def test_and_mode(self) -> None:
"""AND 模式：只有所有因子都看多才买入。"""
from easy_tdx.backtest.combo import FactorSignals, combine_masks
n = 20
# 因子 A: bar 3 买入, bar 8 卖出
fs_a = FactorSignals(
name="A",
buy_mask=np.array([i == 3 for i in range(n)]),
sell_mask=np.array([i == 8 for i in range(n)]),
)
# 因子 B: bar 3 买入, bar 10 卖出
fs_b = FactorSignals(
name="B",
buy_mask=np.array([i == 3 for i in range(n)]),
sell_mask=np.array([i == 10 for i in range(n)]),
)
buy, sell = combine_masks([fs_a, fs_b], mode="AND")
# bar 3 两因子都看多 → AND 买入
assert buy[3] is np.True_
# 其他 bar 不买入
assert int(np.sum(buy)) == 1
# 卖出：只有 bar 8 两因子都看空？不是，A 在 8 卖，B 在 10 卖 → AND 无交集
assert int(np.sum(sell)) == 0
def test_or_mode(self) -> None:
"""OR 模式：任一因子看多就买入。"""
from easy_tdx.backtest.combo import FactorSignals, combine_masks
n = 20
fs_a = FactorSignals(
name="A",
buy_mask=np.array([i == 3 for i in range(n)]),
sell_mask=np.array([i == 8 for i in range(n)]),
)
fs_b = FactorSignals(
name="B",
buy_mask=np.array([i == 5 for i in range(n)]),
sell_mask=np.array([i == 10 for i in range(n)]),
)
buy, sell = combine_masks([fs_a, fs_b], mode="OR")
# bar 3 和 bar 5 都应该买入
assert buy[3] is np.True_
assert buy[5] is np.True_
assert int(np.sum(buy)) == 2
# bar 8 和 bar 10 都应该卖出
assert sell[8] is np.True_
assert sell[10] is np.True_
assert int(np.sum(sell)) == 2
def test_majority_mode_2_factors(self) -> None:
"""MAJORITY 模式（2 因子）：过半 = >1 → 需要两个都同意（等同 AND）。"""
from easy_tdx.backtest.combo import FactorSignals, combine_masks
n = 20
fs_a = FactorSignals(
name="A",
buy_mask=np.array([i == 3 for i in range(n)]),
sell_mask=np.array([i == 8 for i in range(n)]),
)
fs_b = FactorSignals(
name="B",
buy_mask=np.array([i == 3 for i in range(n)]),
sell_mask=np.array([i == 10 for i in range(n)]),
)
buy, sell = combine_masks([fs_a, fs_b], mode="MAJORITY")
# 2 因子 MAJORITY: threshold = 2/2 = 1, need > 1 → need 2 个同意
assert buy[3] is np.True_
assert int(np.sum(buy)) == 1
def test_majority_mode_3_factors(self) -> None:
"""MAJORITY 模式（3 因子）：过半 = >1.5 → 需要至少 2 个同意。"""
from easy_tdx.backtest.combo import FactorSignals, combine_masks
n = 20
fs_a = FactorSignals(
name="A",
buy_mask=np.array([i == 5 for i in range(n)]),
sell_mask=np.array([i == 10 for i in range(n)]),
)
fs_b = FactorSignals(
name="B",
buy_mask=np.array([i == 5 for i in range(n)]),
sell_mask=np.array([i == 12 for i in range(n)]),
)
fs_c = FactorSignals(
name="C",
buy_mask=np.array([i == 7 for i in range(n)]),
sell_mask=np.array([i == 10 for i in range(n)]),
)
buy, sell = combine_masks([fs_a, fs_b, fs_c], mode="MAJORITY")
# bar 5: A 和 B 同意（2/3），满足 > 1.5 → 买入
assert buy[5] is np.True_
# bar 7: 只有 C 同意（1/3），不满足 → 不买入
assert buy[7] is np.False_
assert int(np.sum(buy)) == 1
# bar 10: A 和 C 同意卖出（2/3），满足 → 卖出
assert sell[10] is np.True_
# bar 12: 只有 B 同意（1/3），不满足 → 不卖出
assert sell[12] is np.False_
assert int(np.sum(sell)) == 1
def test_invalid_mode_raises(self) -> None:
"""无效合并模式应抛出 ValueError。"""
from easy_tdx.backtest.combo import FactorSignals, combine_masks
n = 10
fs = FactorSignals(
name="X",
buy_mask=np.zeros(n, dtype=bool),
sell_mask=np.zeros(n, dtype=bool),
)
with pytest.raises(ValueError, match="不支持的合并模式"):
combine_masks([fs, fs], mode="INVALID")
# ── CombinationRunner 测试 ──────────────────────────────────────────────────
class TestCombinationRunner:
"""测试组合回测运行器。"""
def test_run_combination_and(self) -> None:
"""运行 AND 组合：两个确定性策略在第 3 和第 5 bar 都不重叠 → AND 无交易。"""
from easy_tdx.backtest.combo import CombinationRunner
df = _make_df()
runner = CombinationRunner(
strategy_classes=[BuyOnBar3Strategy, BuyOnBar5Strategy],
df=df,
cash=100000.0,
)
result = runner.run_combination(indices=[0, 1], mode="AND")
# BuyOnBar3 在 bar 3 买, BuyOnBar5 在 bar 5 买 → AND 无交集
assert result.performance["total_trades"] == 0
def test_run_combination_or(self) -> None:
"""运行 OR 组合：两个策略买入信号取并集。"""
from easy_tdx.backtest.combo import CombinationRunner
df = _make_df()
runner = CombinationRunner(
strategy_classes=[BuyOnBar3Strategy, BuyOnBar5Strategy],
df=df,
cash=100000.0,
)
result = runner.run_combination(indices=[0, 1], mode="OR")
# OR 模式：bar 3 和 bar 5 都会触发买入信号
# 但由于仓位管理，实际成交可能有限
assert result.performance["total_trades"] >= 1
def test_screen_2_factors(self) -> None:
"""screen 遍历所有 2 因子组合。"""
from easy_tdx.backtest.combo import CombinationRunner
df = _make_df()
runner = CombinationRunner(
strategy_classes=[BuyOnBar3Strategy, BuyOnBar5Strategy, BuyOnBar7Strategy],
df=df,
cash=100000.0,
)
results = runner.screen(combo_sizes=(2,), mode="OR")
# C(3, 2) = 3 组合
assert len(results) == 3
# 每个结果应有名称和绩效
for r in results:
assert " + " in r.name
assert r.result.performance is not None
def test_screen_2_and_3_factors(self) -> None:
"""screen 同时遍历 2 和 3 因子组合。"""
from easy_tdx.backtest.combo import CombinationRunner
df = _make_df()
runner = CombinationRunner(
strategy_classes=[BuyOnBar3Strategy, BuyOnBar5Strategy, BuyOnBar7Strategy],
df=df,
cash=100000.0,
)
results = runner.screen(combo_sizes=(2, 3), mode="OR")
# C(3, 2) + C(3, 3) = 3 + 1 = 4
assert len(results) == 4
# 应按总收益率降序排列
returns = [r.result.performance["total_return"] for r in results]
assert returns == sorted(returns, reverse=True)
def test_screen_with_never_trade(self) -> None:
"""包含 NeverTradeStrategy 的组合不应崩溃。"""
from easy_tdx.backtest.combo import CombinationRunner
df = _make_df()
runner = CombinationRunner(
strategy_classes=[BuyOnBar3Strategy, NeverTradeStrategy],
df=df,
cash=100000.0,
)
results = runner.screen(combo_sizes=(2,), mode="AND", filter_zero_trades=False)
# C(2, 2) = 1，AND 模式下无交易
assert len(results) == 1
assert results[0].result.performance["total_trades"] == 0
def test_screen_with_zero_trade_filtered(self) -> None:
"""screen 默认过滤零交易组合（filter_zero_trades=True）。"""
from easy_tdx.backtest.combo import CombinationRunner
df = _make_df()
runner = CombinationRunner(
strategy_classes=[BuyOnBar3Strategy, BuyOnBar5Strategy, NeverTradeStrategy],
df=df,
cash=100000.0,
)
# AND 模式：大部分组合无交集 → 零交易
results = runner.screen(combo_sizes=(2,), mode="AND", filter_zero_trades=True)
# 所有 AND 组合应该都是零交易（信号不重叠），过滤后为空
assert len(results) == 0