mirror of
https://ghfast.top/https://github.com/aeroxw/tick-stock-panel.git
synced 2026-09-12 16:44:15 +08:00
shy3130
f03bc38a16
策略页全量 run_all 需 ~2 分钟, 期间卡片全空。现在按历史耗时升序执行, 首返时限 (strategy_run_all_first_return_s, 默认 15s) 内算完的策略随 响应返回, 慢策略转后台继续算并逐个写入策略缓存, 前端轮询 cached-summary 逐个点亮卡片数字 (未出的显示脉冲占位)。 后端: - services/strategy_run_queue: 单飞 daemon 工作线程 + handle 状态。 相同 key (资产/周期/日期/策略集) 且未完成的请求搭车现有执行不重算; 已完成的重跑即新执行。后端全局同时只跑一个 run_all, 补上前端 防重入之外的第二道 Numba 并发防线 - run_all 渐进分支 (仅日线 + summary_only): 逐策略增量写缓存 (同日 按 sid 合并), 收尾整体重写保持旧口径; 结果带 computed_at 时间戳; 分钟周期与明细请求保持整段阻塞不变 - 历史耗时落盘 user_data/strategy_run_timings.json, 次日起快策略自动 排前; cached-summary 透传 computed_at 前端: - 请求通用 30s 超时 + 慢接口豁免清单 (run_all/run/backtest/factor 等 300s), 避免一个挂起请求占满 HTTP/1.1 连接拖死全站 - Screener: 收到 pending 后 summaryQuery 每 2s 轮询, 以 computed_at >= started_at 判新 (防同日旧缓存冒充), 8 分钟兜底; StrategyCard 三种尺寸新增 computing 脉冲占位 验证: 新增 13 测试 (排序/落盘/搭车/串行化/端点行为/旧路径兼容), 全量套件 1762 passed; pnpm build 通过。
130 lines
3.7 KiB
Python
130 lines
3.7 KiB
Python
"""strategy_run_queue 单测 — 耗时排序、耗时落盘、handle 状态、单飞管理器。"""
|
|
from __future__ import annotations
|
|
|
|
import threading
|
|
import time
|
|
|
|
from app.services import strategy_run_queue as q
|
|
|
|
|
|
def test_order_strategy_ids_fast_first_unknown_last():
|
|
ids = ["slow_a", "mid_b", "fast_c", "new_d", "new_e"]
|
|
timings = {"slow_a": 5000.0, "fast_c": 100.0, "mid_b": 1000.0}
|
|
assert q.order_strategy_ids(ids, timings) == ["fast_c", "mid_b", "slow_a", "new_d", "new_e"]
|
|
|
|
|
|
def test_run_timings_record_merge_and_load(tmp_path):
|
|
assert q.load_run_timings(tmp_path) == {}
|
|
q.record_run_timings(tmp_path, {"a": 100.0})
|
|
q.record_run_timings(tmp_path, {"b": 200.0, "a": 50.0})
|
|
assert q.load_run_timings(tmp_path) == {"a": 50.0, "b": 200.0}
|
|
|
|
|
|
def test_run_timings_survives_corrupt_file(tmp_path):
|
|
path = q._timings_path(tmp_path)
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text("not json", encoding="utf-8")
|
|
assert q.load_run_timings(tmp_path) == {}
|
|
q.record_run_timings(tmp_path, {"a": 1.0})
|
|
assert q.load_run_timings(tmp_path) == {"a": 1.0}
|
|
|
|
|
|
def test_handle_snapshot_lifecycle():
|
|
h = q.StrategyRunHandle(("k",), ["s1", "s2"])
|
|
snap = h.snapshot()
|
|
assert snap["pending"] == ["s1", "s2"]
|
|
assert snap["done"] is False
|
|
|
|
h.complete("s1", {"total": 3})
|
|
snap = h.snapshot()
|
|
assert snap["results"] == {"s1": {"total": 3}}
|
|
assert snap["pending"] == ["s2"]
|
|
|
|
h.fail("boom")
|
|
snap = h.snapshot()
|
|
assert snap["error"] == "boom"
|
|
assert snap["done"] is True
|
|
|
|
|
|
def _wait_done(handle, timeout=5.0) -> None:
|
|
deadline = time.time() + timeout
|
|
while time.time() < deadline:
|
|
if handle.snapshot()["done"]:
|
|
return
|
|
time.sleep(0.01)
|
|
raise AssertionError("handle 未在时限内完成")
|
|
|
|
|
|
def test_manager_piggybacks_same_running_key():
|
|
mgr = q.StrategyRunManager()
|
|
started = threading.Event()
|
|
release = threading.Event()
|
|
calls: list[int] = []
|
|
|
|
def job(handle):
|
|
calls.append(1)
|
|
started.set()
|
|
assert release.wait(timeout=5)
|
|
|
|
h1 = mgr.get_or_submit(("k",), ["s"], job)
|
|
assert started.wait(timeout=5)
|
|
# 执行中: 相同 key 搭车, 不重复提交
|
|
h2 = mgr.get_or_submit(("k",), ["s"], job)
|
|
assert h2 is h1
|
|
|
|
release.set()
|
|
_wait_done(h1)
|
|
assert calls == [1]
|
|
|
|
# 已完成后: 同 key 再来 → 新执行 (重跑语义)
|
|
h3 = mgr.get_or_submit(("k",), ["s"], job)
|
|
assert h3 is not h1
|
|
_wait_done(h3)
|
|
assert calls == [1, 1]
|
|
|
|
|
|
def test_manager_piggybacks_same_queued_key():
|
|
mgr = q.StrategyRunManager()
|
|
started = threading.Event()
|
|
release = threading.Event()
|
|
|
|
def job(handle):
|
|
started.set()
|
|
assert release.wait(timeout=5)
|
|
|
|
h1 = mgr.get_or_submit(("a",), ["s"], job) # 占住唯一 worker
|
|
assert started.wait(timeout=5)
|
|
# key b 排队 (未开始), 此时 b 的重复请求应搭车排队中的 handle
|
|
h2 = mgr.get_or_submit(("b",), ["s"], job)
|
|
h3 = mgr.get_or_submit(("b",), ["s"], job)
|
|
assert h2 is h3
|
|
|
|
release.set()
|
|
_wait_done(h1)
|
|
_wait_done(h2, timeout=10)
|
|
|
|
|
|
def test_manager_serializes_different_keys():
|
|
mgr = q.StrategyRunManager()
|
|
lock = threading.Lock()
|
|
active: list[str] = []
|
|
overlap: list[list[str]] = []
|
|
|
|
def make_job(name):
|
|
def job(handle):
|
|
with lock:
|
|
active.append(name)
|
|
if len(active) > 1:
|
|
overlap.append(list(active))
|
|
time.sleep(0.1)
|
|
with lock:
|
|
active.remove(name)
|
|
|
|
return job
|
|
|
|
h1 = mgr.get_or_submit(("a",), ["s"], make_job("a"))
|
|
h2 = mgr.get_or_submit(("b",), ["s"], make_job("b"))
|
|
_wait_done(h1, timeout=10)
|
|
_wait_done(h2, timeout=10)
|
|
assert overlap == []
|