mirror of
https://ghfast.top/https://github.com/aeroxw/easy_tdx_max.git
synced 2026-09-12 20:24:19 +08:00
GitHub
4bd5b5d833
- ZIG 右侧突破回补策略：MyTT 新增 ZIG 之字转向（未来函数，含前视偏差警示）； 波谷启动建仓挂硬止损（OCO）→ 见顶清仓记前高 → 右侧突破回补；路径依赖不实现 entry_exit_masks（向量化守护测试白名单）；含寻优预设网格与 --strategy-file 独立文件 - 交易时段感知刷新：realtime/session.py（09:15~11:30:30 / 13:00~15:05）+ GET /market/session；看板 30/60/120s 轮询休市自动暂停（三态状态栏 + 开关持久化 + 手动刷新不受限）；SSE/WS 既有会话语义不动 - 120 分钟 K 线：/bars?category=MIN_120（MAC 原生 Period.MINS×120 优先， 2×60M 相邻聚合兜底，标准客户端上限 400 根）；前端周期选择器同步 - 逐 bar 衍生字段：/bars 与 /bars/index 附带 pre_close/change/change_pct/ amplitude_pct（pre_close≤0.01 兜底防除零） - 159 只核心龙头池：数据资产取自 Fork（东财全行业龙头名单，四组分层）； universe=core 接入 screen scan / SignalScanner / StrengthRanker / market strength； GET /market/core-leaders + WebUI「龙头池」页（搜索/个股详情） - 多 Provider LLM 直连：easy_tdx.ai + /llm/*（DeepSeek/通义/智谱/Kimi/MiniMax/ OpenAI/Claude/Ollama/自定义，openai 兼容 + anthropic 原生双协议）； 配置落盘 ~/.easy_tdx/llm.json（WebUI「AI 设置」页 ⇆ 手工编辑双向兼容， 文件>环境变量>预设；key 脱敏回显/CLEAR 清除）；「AI 解读」后台任务化 （复用 task_runner，提交+轮询，不占 HTTP 连接）；思考型模型空白正文防御 （reasoning_content 耗尽 max_tokens → 可操作报错；默认 16000）； AI 解读历史页（自动归档 Prompt/正文/策略上下文 + 去回测带参引导） - WebUI 加固：SPA fallback 对未知 /api/* 返回 JSON 404（不再 200 HTML 伪装解析错）； index.html 一律 Cache-Control: no-store（防缓存旧资源引用）；路由兜底重定向； 全局风险提示常驻底栏 + 龙头池/AI 解读针对性免责声明 - 测试：新增 9 个单测文件共 59 例；黄金基线仅新增 zig_breakout 条目（其余零漂移）； 全量 1448 例通过
98 lines
3.4 KiB
Python
98 lines
3.4 KiB
Python
"""zig_breakout 内置策略单元测试（借鉴 Fork 移植，v1.29）。
|
||
|
||
覆盖：注册表登记与参数 schema、合成锯齿行情能产生交易、
|
||
止损单挂在买入信号上（OCO bracket）、寻优预设网格登记。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import numpy as np
|
||
import pandas as pd
|
||
import pytest
|
||
|
||
from easy_tdx.backtest.engine import BacktestEngine
|
||
from easy_tdx.backtest.strategies import get_registry
|
||
from easy_tdx.backtest.strategies.presets import STRATEGY_PRESETS
|
||
|
||
|
||
def _zigzag_df(n: int = 300, seed: int = 42) -> pd.DataFrame:
|
||
"""先跌后大涨再回调的合成行情（触发 ZIG 波谷启动与见顶清仓）。"""
|
||
rng = np.random.default_rng(seed)
|
||
trend = np.concatenate(
|
||
[
|
||
np.linspace(100, 80, n // 3),
|
||
np.linspace(80, 130, n * 2 // 5),
|
||
np.linspace(130, 110, n - n // 3 - n * 2 // 5),
|
||
]
|
||
)
|
||
close = trend + rng.normal(0, 0.8, len(trend))
|
||
high = close + rng.uniform(0, 1.5, len(trend))
|
||
low = close - rng.uniform(0, 1.5, len(trend))
|
||
return pd.DataFrame(
|
||
{
|
||
"datetime": pd.date_range("2024-01-01", periods=len(trend), freq="B"),
|
||
"open": close + rng.normal(0, 0.3, len(trend)),
|
||
"high": high,
|
||
"low": low,
|
||
"close": close,
|
||
"vol": rng.integers(1e6, 5e6, len(trend)).astype(float),
|
||
"amount": close * 1e6,
|
||
}
|
||
)
|
||
|
||
|
||
def test_registry_entry_and_params():
|
||
entry = get_registry().get("zig_breakout")
|
||
assert entry.label == "ZIG 右侧突破回补"
|
||
names = [p.name for p in entry.params]
|
||
assert names == ["zig_delta", "confirm_pct", "hhv_period", "stop_loss_pct"]
|
||
defaults = {p.name: p.default for p in entry.params}
|
||
assert defaults == {
|
||
"zig_delta": 10.0,
|
||
"confirm_pct": 2.0,
|
||
"hhv_period": 20,
|
||
"stop_loss_pct": 3.0,
|
||
}
|
||
|
||
|
||
def test_build_validates_params():
|
||
entry = get_registry().get("zig_breakout")
|
||
inst = entry.build({"zig_delta": 5})
|
||
assert inst.p["zig_delta"] == 5.0 and inst.p["hhv_period"] == 20
|
||
with pytest.raises(ValueError):
|
||
entry.build({"zig_delta": -1}) # 低于 min_value
|
||
|
||
|
||
def test_strategy_trades_and_bracket_stop():
|
||
entry = get_registry().get("zig_breakout")
|
||
result = BacktestEngine(entry.build(), cash=1_000_000).run(_zigzag_df())
|
||
assert len(result.trades) > 0
|
||
# 锯齿行情应至少出现一次 BUY（trades 为 DataFrame）
|
||
assert (result.trades["direction"] == "BUY").any()
|
||
assert (result.trades["direction"] == "SELL").any()
|
||
|
||
|
||
def test_strategy_file_variant_loadable():
|
||
"""strategies/zig_breakout.py 独立文件可供 --strategy-file 加载。"""
|
||
import importlib.util
|
||
from pathlib import Path
|
||
|
||
path = Path(__file__).resolve().parents[2] / "strategies" / "zig_breakout.py"
|
||
spec = importlib.util.spec_from_file_location("zig_file_test", path)
|
||
mod = importlib.util.module_from_spec(spec)
|
||
spec.loader.exec_module(mod)
|
||
result = BacktestEngine(mod.ZigBreakoutStrategy(), cash=1_000_000).run(_zigzag_df())
|
||
assert len(result.trades) > 0
|
||
assert (result.trades["direction"] == "BUY").any()
|
||
|
||
|
||
def test_preset_grid_registered():
|
||
assert "zig_breakout" in STRATEGY_PRESETS
|
||
grid = STRATEGY_PRESETS["zig_breakout"]
|
||
assert "zig_delta" in grid and "confirm_pct" in grid
|
||
# 笛卡尔积不超过寻优器上限
|
||
n = 1
|
||
for vals in grid.values():
|
||
n *= len(vals)
|
||
assert n <= 200
|