mirror of
https://ghfast.top/https://github.com/aeroxw/tick-stock-panel.git
synced 2026-09-12 21:24:16 +08:00
lytem28
b6cf0495e1
Unify strategy execution across backtest, screener, and monitoring; isolate backtest workloads in spawn workers; and add shared matrix caching plus valid-bar indicator acceleration.
362 lines
12 KiB
Python
362 lines
12 KiB
Python
"""优化器编排测试 — 用假 service 注入受控 stats, 验证排序/取消/进度/目标方向。"""
|
|
from __future__ import annotations
|
|
|
|
import threading
|
|
from dataclasses import dataclass
|
|
from datetime import date
|
|
|
|
import pytest
|
|
|
|
from app.backtest.optimizer import OptimizeConfig, StrategyOptimizer
|
|
|
|
# ---- 假 StrategyDef / 引擎 / service ----
|
|
|
|
@dataclass
|
|
class _FakeDef:
|
|
meta: dict
|
|
execution_backend: str = "polars_expr"
|
|
|
|
|
|
class _FakeEngine:
|
|
def __init__(self, params_meta, execution_backend="polars_expr"):
|
|
self._def = _FakeDef(
|
|
meta={"params": params_meta},
|
|
execution_backend=execution_backend,
|
|
)
|
|
|
|
def get(self, strategy_id):
|
|
return self._def
|
|
|
|
|
|
@dataclass
|
|
class _FakeResult:
|
|
stats: dict
|
|
error: str | None = None
|
|
|
|
|
|
class _FakeCache:
|
|
def __init__(self):
|
|
self.closed = False
|
|
|
|
def snapshot(self):
|
|
return {"current_bytes": 64, "hits": 2}
|
|
|
|
def close(self):
|
|
self.closed = True
|
|
|
|
|
|
class _FakeService:
|
|
"""run() 依据 params 返回受控 stats: sortino = ma_proximity 的映射, 便于校验排序。"""
|
|
|
|
def __init__(self, score_fn):
|
|
self.score_fn = score_fn
|
|
self.calls = []
|
|
self.prepared_calls = []
|
|
self.prepared_values = []
|
|
self.result_policies = []
|
|
self.matrix_cache_max_bytes = None
|
|
self.cache = _FakeCache()
|
|
self._lock = threading.Lock()
|
|
|
|
def prepare_matrix_optimization(self, configs, *, matrix_cache_max_bytes):
|
|
self.prepared_calls.append(configs)
|
|
self.matrix_cache_max_bytes = matrix_cache_max_bytes
|
|
return type("Prepared", (), {
|
|
"market_data": type("Market", (), {"nbytes": 1234})(),
|
|
"compute_cache": self.cache,
|
|
})()
|
|
|
|
def run(
|
|
self,
|
|
config,
|
|
progress_cb=None,
|
|
cancel_event=None,
|
|
prepared=None,
|
|
result_policy=None,
|
|
):
|
|
with self._lock:
|
|
self.calls.append(dict(config.params or {}))
|
|
self.prepared_values.append(prepared)
|
|
self.result_policies.append(result_policy)
|
|
return self.score_fn(config.params or {})
|
|
|
|
|
|
PARAMS_META = [
|
|
{"id": "ma_proximity", "type": "float", "default": 0.02, "min": 0.01, "max": 0.05, "step": 0.005},
|
|
]
|
|
|
|
|
|
def _optimizer(score_fn, execution_backend="polars_expr"):
|
|
return StrategyOptimizer(
|
|
_FakeService(score_fn),
|
|
_FakeEngine(PARAMS_META, execution_backend=execution_backend),
|
|
)
|
|
|
|
|
|
def _cfg(**kw):
|
|
base = dict(
|
|
strategy_id="s", symbols=None, start=date(2024, 1, 1), end=date(2024, 6, 1),
|
|
param_grid={"ma_proximity": [0.01, 0.02, 0.03]}, objective="sortino", max_workers=4,
|
|
)
|
|
base.update(kw)
|
|
return OptimizeConfig(**base)
|
|
|
|
|
|
def test_ranks_best_by_objective_max():
|
|
# sortino 随 ma_proximity 递增 -> 最大值应为 0.03
|
|
def score(p):
|
|
return _FakeResult(stats={"sortino": p["ma_proximity"] * 100})
|
|
out = _optimizer(score).optimize(_cfg())
|
|
assert out["best_params"] == {"ma_proximity": 0.03}
|
|
assert out["best_score"] == 3.0
|
|
assert out["n_combinations"] == 3
|
|
assert out["n_completed"] == 3
|
|
assert [r["rank"] for r in out["results"]] == [1, 2, 3]
|
|
assert out["results"][0]["params"] == {"ma_proximity": 0.03}
|
|
|
|
|
|
def test_all_combos_executed_once():
|
|
def score(p):
|
|
return _FakeResult(stats={"sortino": 1.0})
|
|
opt = _optimizer(score)
|
|
out = opt.optimize(_cfg(param_grid={"ma_proximity": [0.01, 0.02, 0.03, 0.04, 0.05]}))
|
|
assert out["n_combinations"] == 5
|
|
# 每组恰跑一次
|
|
ran = sorted(
|
|
call["ma_proximity"]
|
|
for call, policy in zip(opt.service.calls, opt.service.result_policies, strict=True)
|
|
if policy is not None
|
|
)
|
|
assert ran == [0.01, 0.02, 0.03, 0.04, 0.05]
|
|
assert opt.service.result_policies.count(None) == 1
|
|
|
|
|
|
def test_min_direction_objective_restores_display_sign():
|
|
# avg_holding_days 是 min 方向: 最小者最优, 且 best_score 必须是原始正值 (非内部取负值)
|
|
def score(p):
|
|
return _FakeResult(stats={"avg_holding_days": p["ma_proximity"] * 100})
|
|
out = _optimizer(score).optimize(_cfg(objective="avg_holding_days"))
|
|
assert out["best_params"] == {"ma_proximity": 0.01}
|
|
# min 方向: 最优 avg_holding_days = 0.01*100 = 1.0, 用户应看到 +1.0 而非 -1.0
|
|
assert out["best_score"] == 1.0
|
|
# results 不应外露内部排序键 _sort
|
|
assert all("_sort" not in r for r in out["results"])
|
|
assert out["results"][0]["objective_raw"] == 1.0
|
|
|
|
|
|
def test_max_drawdown_objective_prefers_smaller_drawdown():
|
|
# max_drawdown 为负值, max 方向: -0.1 (回撤更小) 应优于 -0.3
|
|
def score(p):
|
|
dd = {0.01: -0.1, 0.02: -0.3, 0.03: -0.2}[p["ma_proximity"]]
|
|
return _FakeResult(stats={"max_drawdown": dd})
|
|
out = _optimizer(score).optimize(_cfg(objective="max_drawdown"))
|
|
assert out["best_params"] == {"ma_proximity": 0.01}
|
|
assert out["best_score"] == -0.1 # 展示原始负值
|
|
|
|
|
|
def test_service_exception_isolated_not_crashing_batch():
|
|
# 某组 service.run 抛异常 -> 应记为该组失败, 其余组正常完成, 不拖垮整批
|
|
def score(p):
|
|
if p["ma_proximity"] == 0.02:
|
|
raise KeyError("boom")
|
|
return _FakeResult(stats={"sortino": p["ma_proximity"] * 100})
|
|
out = _optimizer(score).optimize(_cfg())
|
|
assert out["n_completed"] == 3 # 三组都有结果记录 (含失败组)
|
|
assert out["best_params"] == {"ma_proximity": 0.03} # 最优组不受影响
|
|
failed = [r for r in out["results"] if r.get("error")]
|
|
assert len(failed) == 1
|
|
assert "boom" in failed[0]["error"]
|
|
|
|
|
|
def test_backtest_kwargs_illegal_key_rejected():
|
|
def score(p):
|
|
return _FakeResult(stats={"sortino": 1.0})
|
|
with pytest.raises(ValueError, match=r"非法字段|不能包含"):
|
|
_optimizer(score).optimize(_cfg(backtest_kwargs={"bad_field": 1}))
|
|
|
|
|
|
def test_backtest_kwargs_reserved_key_rejected():
|
|
def score(p):
|
|
return _FakeResult(stats={"sortino": 1.0})
|
|
with pytest.raises(ValueError, match="不能包含"):
|
|
_optimizer(score).optimize(_cfg(backtest_kwargs={"symbols": ["x"]}))
|
|
|
|
|
|
def test_base_params_merged_and_overridden_by_sweep():
|
|
# base_params 提供固定参数, combo 覆盖同名; 记录 service 实际收到的 params
|
|
def score(p):
|
|
return _FakeResult(stats={"sortino": 1.0})
|
|
opt = _optimizer(score)
|
|
opt.optimize(_cfg(base_params={"ma_proximity": 0.99, "other": 7}))
|
|
# 每次 run 收到的 params: ma_proximity 被 combo 覆盖, other 保留
|
|
for call, policy in zip(opt.service.calls, opt.service.result_policies, strict=True):
|
|
if policy is None:
|
|
continue
|
|
assert call["other"] == 7
|
|
assert call["ma_proximity"] in (0.01, 0.02, 0.03)
|
|
|
|
|
|
def test_none_and_error_results_sink_to_bottom():
|
|
# ma_proximity=0.02 的组返回 error, 0.03 的 sortino=None -> 都应排在有效结果之后
|
|
def score(p):
|
|
if p["ma_proximity"] == 0.02:
|
|
return _FakeResult(stats={}, error="boom")
|
|
if p["ma_proximity"] == 0.03:
|
|
return _FakeResult(stats={"sortino": None})
|
|
return _FakeResult(stats={"sortino": 5.0})
|
|
out = _optimizer(score).optimize(_cfg())
|
|
assert out["best_params"] == {"ma_proximity": 0.01}
|
|
assert out["best_score"] == 5.0
|
|
# 失败/None 组仍在结果里但 rank 靠后
|
|
assert out["n_completed"] == 3
|
|
assert out["results"][0]["params"] == {"ma_proximity": 0.01}
|
|
|
|
|
|
def test_cancel_event_stops_remaining():
|
|
ev = threading.Event()
|
|
ev.set() # 一开始就取消
|
|
|
|
def score(p):
|
|
return _FakeResult(stats={"sortino": 1.0})
|
|
opt = _optimizer(score)
|
|
out = opt.optimize(_cfg(), cancel_event=ev)
|
|
# 取消后所有组跳过 -> 无有效结果
|
|
assert opt.service.calls == []
|
|
assert out["best_params"] is None
|
|
|
|
|
|
def test_progress_callback_reports_done_total():
|
|
seen = []
|
|
|
|
def score(p):
|
|
return _FakeResult(stats={"sortino": 1.0})
|
|
|
|
def cb(msg):
|
|
seen.append(msg)
|
|
_optimizer(score).optimize(_cfg(), progress_cb=cb)
|
|
assert len(seen) == 4
|
|
assert seen[-1]["type"] == "optimizer_finalize"
|
|
assert seen[-1]["done"] == 3
|
|
assert all(m["total"] == 3 for m in seen)
|
|
|
|
|
|
def test_matrix_optimizer_prepares_once_and_reuses_same_market_data():
|
|
def score(p):
|
|
return _FakeResult(stats={"sortino": p["ma_proximity"]})
|
|
|
|
opt = _optimizer(score, execution_backend="matrix_native")
|
|
out = opt.optimize(_cfg(max_workers=8))
|
|
|
|
assert len(opt.service.prepared_calls) == 1
|
|
assert len(opt.service.prepared_calls[0]) == 3
|
|
assert len({id(value) for value in opt.service.prepared_values}) == 1
|
|
assert opt.service.prepared_values[0] is not None
|
|
assert out["requested_max_workers"] == 8
|
|
assert out["effective_workers"] == 1
|
|
assert out["shared_market_data"] is True
|
|
assert out["shared_market_data_bytes"] == 1234
|
|
assert out["best_backtest"] is not None
|
|
assert out["matrix_compute_cache"]["released"] is True
|
|
assert opt.service.cache.closed is True
|
|
assert opt.service.matrix_cache_max_bytes == 512 * 1024 * 1024
|
|
|
|
|
|
def test_optimizer_trial_policy_skips_mc_but_best_backtest_is_full():
|
|
def score(p):
|
|
return _FakeResult(stats={"sortino": p["ma_proximity"]})
|
|
|
|
opt = _optimizer(score)
|
|
opt.optimize(_cfg())
|
|
|
|
trial_policies = [policy for policy in opt.service.result_policies if policy is not None]
|
|
assert trial_policies
|
|
assert all(policy.include_monte_carlo is False for policy in trial_policies)
|
|
assert opt.service.result_policies[-1] is None
|
|
|
|
|
|
def test_mc_objective_trial_policy_keeps_monte_carlo():
|
|
def score(p):
|
|
return _FakeResult(stats={"mc_maxdd_p95": -p["ma_proximity"]})
|
|
|
|
opt = _optimizer(score)
|
|
opt.optimize(_cfg(objective="mc_maxdd_p95"))
|
|
trial_policies = [policy for policy in opt.service.result_policies if policy is not None]
|
|
assert all(policy.include_monte_carlo is True for policy in trial_policies)
|
|
|
|
|
|
def test_missing_objective_is_an_explicit_trial_error():
|
|
def score(p):
|
|
return _FakeResult(stats={"sharpe": 1.0})
|
|
|
|
out = _optimizer(score).optimize(_cfg())
|
|
assert out["best_params"] is None
|
|
assert all("缺少优化目标字段" in row["error"] for row in out["results"])
|
|
|
|
|
|
def test_matrix_cache_closes_when_progress_callback_raises_after_prepare():
|
|
def score(p):
|
|
return _FakeResult(stats={"sortino": 1.0})
|
|
|
|
opt = _optimizer(score, execution_backend="matrix_native")
|
|
|
|
def fail_on_prepare(message):
|
|
if message["type"] == "optimizer_prepare":
|
|
raise RuntimeError("progress failed")
|
|
|
|
with pytest.raises(RuntimeError, match="progress failed"):
|
|
opt.optimize(_cfg(), progress_cb=fail_on_prepare)
|
|
assert opt.service.cache.closed is True
|
|
|
|
|
|
def test_matrix_cache_closes_when_cancelled_after_first_trial():
|
|
event = threading.Event()
|
|
|
|
def score(p):
|
|
event.set()
|
|
return _FakeResult(stats={"sortino": 1.0})
|
|
|
|
opt = _optimizer(score, execution_backend="matrix_native")
|
|
out = opt.optimize(_cfg(), cancel_event=event)
|
|
|
|
assert out["n_completed"] == 1
|
|
assert out["best_backtest"] is None
|
|
assert opt.service.cache.closed is True
|
|
|
|
|
|
def test_matrix_cache_closes_when_best_backtest_raises():
|
|
call_count = 0
|
|
|
|
def score(p):
|
|
nonlocal call_count
|
|
call_count += 1
|
|
if call_count == 4:
|
|
raise RuntimeError("final failed")
|
|
return _FakeResult(stats={"sortino": p["ma_proximity"]})
|
|
|
|
opt = _optimizer(score, execution_backend="matrix_native")
|
|
with pytest.raises(RuntimeError, match="final failed"):
|
|
opt.optimize(_cfg())
|
|
assert opt.service.cache.closed is True
|
|
|
|
|
|
def test_cancelled_matrix_optimizer_skips_expensive_preparation():
|
|
event = threading.Event()
|
|
event.set()
|
|
|
|
def score(p):
|
|
return _FakeResult(stats={"sortino": 1.0})
|
|
|
|
opt = _optimizer(score, execution_backend="matrix_native")
|
|
out = opt.optimize(_cfg(), cancel_event=event)
|
|
|
|
assert opt.service.prepared_calls == []
|
|
assert opt.service.calls == []
|
|
assert out["n_completed"] == 0
|
|
|
|
|
|
def test_invalid_objective_rejected():
|
|
def score(p):
|
|
return _FakeResult(stats={"sortino": 1.0})
|
|
with pytest.raises(ValueError, match="不支持的优化目标"):
|
|
_optimizer(score).optimize(_cfg(objective="not_a_metric"))
|