mirror of
https://ghfast.top/https://github.com/aeroxw/easy_tdx_max.git
synced 2026-09-12 13:24:18 +08:00
Justin Gu
e374a0da28
对 v1.21→v1.32.5 的 249 文件 4.2 万行改动做六路专项审查，本轮落地全部发现： 回测正确性：组合收益 fillna(0) 虚增、轮动停牌日过期价成交、单标的 WF 逐窗指标 被预热区稀释（三件套均带先红后绿回归）；worst_drawdown 方向、grading 容错、 组合体检品种费率、寻优端点费率透传。 安全：LLM api_url 仅 http/https 且禁 userinfo（封死 file:// 读取与 Key 外送链）、 错误响应不回显原始 body、响应体 2MB 上限、配置原子写、坏配置字段级防御。 数据：涨跌停价整数分币舍入（67/318/90 个价位错 1 分漏判清零）、交易时段/采样/ provisional 统一沪时区、warehouse 增量缺口自动全量重拉、provisional 定点转正、 baostock 真故障抛错 + W/M 去 tradestatus（实测服务端报错，周月兜底此前从未工作） + 指数 vol 股→手（实测锚定）、ccpm 结构变更抛错。 Web API：缓存键补 count/vipdoc、NaN 清洗先于缓存、count>800 分页取全量、 submit 透传真实状态、pending 不再被淘汰成幽灵、watchlist/server 入参约束。 公式：FILTER 去副作用、0-1 值域误判收严、递归深度上限、REF 负移位显式禁止。 前端：4 处请求竞态序号守卫、Sparkline viewBox、北交所 market=2 映射、 空数据缓存死角、AI 弹窗卸载中止轮询、量能/资金日历口径修正。 CLI/CI：warehouse sync 失败 exit 1、参数校验干净报错、release 真实发布 SHA256、 CI 超时与缓存、spec 补 baostock 前提。 约 60 条回归测试先红后绿；pytest 1820 全过，ruff/mypy/vue-tsc/node --test 全绿。
293 lines
9.8 KiB
Python
293 lines
9.8 KiB
Python
"""通达信公式解析器测试（tokenizer / AST / 白名单求值 / 信号归类 / 安全性）。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import numpy as np
|
||
import pandas as pd
|
||
import pytest
|
||
|
||
from easy_tdx.formula import FormulaError, compile_formula
|
||
|
||
|
||
def _df(n: int = 60, seed: int = 3) -> pd.DataFrame:
|
||
rng = np.random.default_rng(seed)
|
||
dates = pd.date_range("2024-01-01", periods=n, freq="B")
|
||
close = 10.0 * np.cumprod(1.0 + 0.002 + rng.normal(0, 0.015, n))
|
||
return pd.DataFrame(
|
||
{
|
||
"datetime": dates,
|
||
"open": close * 0.999,
|
||
"high": close * 1.02,
|
||
"low": close * 0.98,
|
||
"close": close,
|
||
"vol": rng.uniform(1e6, 5e6, n),
|
||
"amount": close * rng.uniform(1e6, 5e6, n),
|
||
}
|
||
)
|
||
|
||
|
||
# ── 编译与语法 ───────────────────────────────────────────────────────────────
|
||
|
||
|
||
def test_compile_and_outputs():
|
||
formula = compile_formula(
|
||
"""
|
||
N := 9;
|
||
RSV := (C - LLV(L, N)) / (HHV(H, N) - LLV(L, N)) * 100;
|
||
K := SMA(RSV, 3, 1);
|
||
金叉: CROSS(K, 20);
|
||
强度: K;
|
||
"""
|
||
)
|
||
res = formula.compute(_df())
|
||
assert "金叉" in res.signals
|
||
assert "强度" in res.values
|
||
frame = res.to_frame()
|
||
assert list(frame.columns) == ["金叉", "强度"]
|
||
assert len(frame) == 60
|
||
|
||
|
||
def test_syntax_error_has_position():
|
||
with pytest.raises(FormulaError):
|
||
compile_formula("A := ;")
|
||
with pytest.raises(FormulaError):
|
||
compile_formula("A := UNKNOWN_FUNC(C)")
|
||
with pytest.raises(FormulaError):
|
||
compile_formula("A := B + ") # 引用未定义变量且语法断裂
|
||
|
||
|
||
def test_unknown_variable_rejected():
|
||
with pytest.raises(FormulaError, match="未知变量"):
|
||
compile_formula("A: X1;").compute(_df())
|
||
|
||
|
||
def test_unknown_function_rejected():
|
||
with pytest.raises(FormulaError, match="白名单"):
|
||
compile_formula("A: EVAL(C);").compute(_df())
|
||
|
||
|
||
def test_empty_formula_rejected():
|
||
with pytest.raises(FormulaError, match="为空"):
|
||
compile_formula("{只有注释}")
|
||
|
||
|
||
def test_no_python_eval_injection():
|
||
"""公式层不走 Python eval：危险标识符按未知变量/函数拒绝。"""
|
||
with pytest.raises(FormulaError):
|
||
compile_formula("__import__('os'): 1;").compute(_df())
|
||
with pytest.raises(FormulaError):
|
||
compile_formula("A: OPEN(C);").compute(_df())
|
||
|
||
|
||
# ── 语义正确性 ───────────────────────────────────────────────────────────────
|
||
|
||
|
||
def test_series_aliases():
|
||
"""C/O/H/L/V/AMOUNT 别名与底层列一致。"""
|
||
df = _df(50)
|
||
res = compile_formula("高价: H; 低价: L; 收盘: CLOSE; 量: VOL; 额: AMOUNT;").compute(df)
|
||
assert np.allclose(res.columns["高价"], df["high"])
|
||
assert np.allclose(res.columns["收盘"], df["close"])
|
||
assert np.allclose(res.columns["量"], df["vol"])
|
||
|
||
|
||
def test_ma_matches_mytt():
|
||
from easy_tdx.MyTT import MA
|
||
|
||
df = _df(50)
|
||
res = compile_formula("均线: MA(C, 5);").compute(df)
|
||
assert np.allclose(res.columns["均线"], MA(df["close"].to_numpy(), 5), equal_nan=True)
|
||
|
||
|
||
def test_cross_semantics():
|
||
"""CROSS(A,B)：A 上穿 B 的那一根为 1，其余 0。"""
|
||
df = _df(50)
|
||
res = compile_formula(
|
||
"""
|
||
快: MA(C, 3);
|
||
慢: MA(C, 10);
|
||
金叉: CROSS(快, 慢);
|
||
"""
|
||
).compute(df)
|
||
golden = res.columns["金叉"]
|
||
assert set(np.unique(golden[np.isfinite(golden)])).issubset({0.0, 1.0})
|
||
assert golden.sum() >= 0 # 结构完整（趋势数据至少存在或为 0）
|
||
# CROSS 手工复算对拍
|
||
from easy_tdx.MyTT import CROSS, MA
|
||
|
||
fast = MA(df["close"].to_numpy(), 3)
|
||
slow = MA(df["close"].to_numpy(), 10)
|
||
assert np.allclose(golden, CROSS(fast, slow), equal_nan=True)
|
||
|
||
|
||
def test_safe_division_zero_denominator_nan():
|
||
"""除零 → NaN（不炸、不 inf）。"""
|
||
df = _df(30)
|
||
res = compile_formula("比值: C / (C - C);").compute(df) # 分母全 0
|
||
assert np.isnan(res.columns["比值"]).all()
|
||
|
||
|
||
def test_logic_operators():
|
||
df = _df(40)
|
||
res = compile_formula(
|
||
"""
|
||
条件1: C > MA(C, 5);
|
||
条件2: C > MA(C, 20);
|
||
同时: 条件1 AND 条件2;
|
||
任一: 条件1 OR 条件2;
|
||
取反: NOT(条件1);
|
||
"""
|
||
).compute(df)
|
||
c1 = res.columns["条件1"] > 0.5
|
||
c2 = res.columns["条件2"] > 0.5
|
||
assert np.allclose(res.columns["同时"] > 0.5, c1 & c2)
|
||
assert np.allclose(res.columns["任一"] > 0.5, c1 | c2)
|
||
assert np.allclose(res.columns["取反"] > 0.5, ~c1)
|
||
|
||
|
||
def test_comparison_and_unary():
|
||
df = _df(30)
|
||
res = compile_formula("跌幅: -(C - REF(C, 1)) / REF(C, 1) * 100; 平: C == C;").compute(df)
|
||
assert (res.columns["平"] == 1.0).all()
|
||
assert "跌幅" in res.values
|
||
|
||
|
||
def test_warmup_nan_not_signal():
|
||
"""预热期 NaN 不产生信号（比较含 NaN → 0）。"""
|
||
df = _df(30)
|
||
res = compile_formula("信号: CROSS(MA(C, 20), MA(C, 25));").compute(df)
|
||
sig = res.columns["信号"]
|
||
assert np.nanmax(np.nan_to_num(sig[:25])) <= 1.0
|
||
assert np.isnan(sig).sum() == 0 # 布尔输出不含 NaN
|
||
|
||
|
||
def test_output_classification_boolean_vs_numeric():
|
||
"""比较/逻辑输出 → 信号；数值输出 → 数值列；0/1 值域数值也归信号。"""
|
||
df = _df(40)
|
||
res = compile_formula(
|
||
"""
|
||
布尔输出: C > REF(C, 1);
|
||
数值输出: MA(C, 5) - MA(C, 20);
|
||
"""
|
||
).compute(df)
|
||
assert res.signals == ["布尔输出"]
|
||
assert res.values == ["数值输出"]
|
||
|
||
|
||
def test_last_row_for_screening():
|
||
df = _df(30)
|
||
res = compile_formula("买入: CROSS(MA(C, 3), MA(C, 10)); 值: MA(C, 5);").compute(df)
|
||
last = res.last_row()
|
||
assert set(last) == {"买入", "值"}
|
||
assert last["买入"] in (0.0, 1.0)
|
||
|
||
|
||
def test_chinese_identifier_and_comment():
|
||
df = _df(30)
|
||
formula = compile_formula(
|
||
"""
|
||
{这是注释：N 周期}
|
||
周期 := 5;
|
||
均线: MA(C, 周期);
|
||
"""
|
||
)
|
||
res = formula.compute(df)
|
||
assert res.columns["均线"][0] != res.columns["均线"][-1]
|
||
|
||
|
||
def test_compiled_formula_reusable_across_frames():
|
||
f = compile_formula("值: MA(C, 5);")
|
||
r1 = f.compute(_df(30, seed=1))
|
||
r2 = f.compute(_df(40, seed=2))
|
||
assert len(r1.columns["值"]) == 30
|
||
assert len(r2.columns["值"]) == 40
|
||
|
||
|
||
def test_compiled_formula_is_dataclass_safe():
|
||
"""CompiledFormula 可 pickle（进程池/后台任务传输）。"""
|
||
import pickle
|
||
|
||
f = compile_formula("值: MA(C, 5);")
|
||
f2 = pickle.loads(pickle.dumps(f))
|
||
assert np.allclose(
|
||
f.compute(_df(20)).columns["值"], f2.compute(_df(20)).columns["值"], equal_nan=True
|
||
)
|
||
|
||
|
||
# ── 回归：归类收严 / FILTER 副作用 / 递归上限 / REF 负移位（审查修复） ────────
|
||
|
||
|
||
def test_ratio_output_classified_as_value_not_signal():
|
||
"""0~1 区间的价格比率是数值列，不是信号列（旧码按 [0,1] 值域兜底误判）。"""
|
||
res = compile_formula("比率: C / HHV(C, 20);").compute(_df(40))
|
||
assert res.signals == []
|
||
assert res.values == ["比率"]
|
||
|
||
|
||
def test_normalized_oscillator_classified_as_value():
|
||
"""归一化振荡器（RSI/100）是数值列，不是信号列。"""
|
||
res = compile_formula("强度: RSI(C, 14) / 100;").compute(_df(40))
|
||
assert res.signals == []
|
||
assert res.values == ["强度"]
|
||
|
||
|
||
def test_binary_zero_one_values_still_signal():
|
||
"""真正的 0/1 两值输出仍兜底归信号列。"""
|
||
res = compile_formula("X: IF(C > MA(C, 5), 1, 0);").compute(_df(40))
|
||
assert res.signals == ["X"]
|
||
|
||
|
||
def test_filter_does_not_pollute_series():
|
||
"""FILTER(C, N) 不改写输入序列：同公式后续 MA(C, 2) 与未过滤一致。"""
|
||
df = _df(30)
|
||
res = compile_formula("A: FILTER(C, 2); B: MA(C, 2);").compute(df)
|
||
close = pd.to_numeric(df["close"]).to_numpy(dtype=float)
|
||
expected = pd.Series(close).rolling(2).mean().to_numpy()
|
||
np.testing.assert_allclose(res.columns["B"], expected, equal_nan=True)
|
||
|
||
|
||
def test_deep_paren_nesting_formula_error():
|
||
"""超深括号嵌套抛 FormulaError（嵌套过深），而非 RecursionError 逃逸。"""
|
||
text = "X: " + "(" * 5000 + "C" + ")" * 5000 + ";"
|
||
with pytest.raises(FormulaError, match="嵌套过深"):
|
||
compile_formula(text)
|
||
|
||
|
||
def test_deep_unary_chain_formula_error():
|
||
"""超长一元运算符链同样受深度上限保护。"""
|
||
with pytest.raises(FormulaError, match="嵌套过深"):
|
||
compile_formula("X: " + "!" * 5000 + "C;")
|
||
|
||
|
||
def test_moderate_nesting_still_compiles():
|
||
"""常规嵌套深度不受上限影响。"""
|
||
res = compile_formula("X: -(-(-(C + 1) * 2) + 3);").compute(_df(10))
|
||
assert res.columns["X"].shape == (10,)
|
||
|
||
|
||
def test_ref_negative_shift_banned():
|
||
"""REF 负移位（未来函数）显式 FormulaError，不再依赖 float 类型巧合。"""
|
||
with pytest.raises(FormulaError, match="负移位"):
|
||
compile_formula("X: REF(C, -1);").compute(_df(30))
|
||
|
||
|
||
def test_ref_negative_via_expression_banned():
|
||
"""负移位经表达式算出（如 0-1）同样被禁。"""
|
||
with pytest.raises(FormulaError, match="负移位"):
|
||
compile_formula("X: REF(C, 0 - 1);").compute(_df(30))
|
||
|
||
|
||
def test_ref_positive_still_works():
|
||
res = compile_formula("X: REF(C, 1);").compute(_df(30))
|
||
assert np.isnan(res.columns["X"][0])
|
||
assert res.columns["X"][1] == pytest.approx(float(pd.to_numeric(_df(30)["close"]).iloc[0]))
|
||
|
||
|
||
def test_mytt_internal_negative_ref_unaffected():
|
||
"""MyTT 库内直调（ICHIMOKU 迟行带）不经公式白名单，负移位仍可用。"""
|
||
from easy_tdx.MyTT import REF
|
||
|
||
close = np.arange(5, dtype=float)
|
||
out = REF(close, -1)
|
||
assert out[0] == pytest.approx(1.0)
|