fix(portfolio): provider download加threading超时防御+Layer1链路闭环
xtdata.download_financial_data 阻塞无超时,休市/服务不响应卡死整个回测 (实证:周六休市HS300全成分首次download卡死,已缓存股票0.5s)。 包threading+join(120s)超时跳过读缓存(数据不全但回测不卡死)。 Layer1 MVP链路完整闭环:runner→BulletTrade engine→AllWeather策略→miniQMT→JSON (7.49%收益/37交易日/选股5只ETF/净值37点/指标total_return+sharpe+max_drawdown全有)。 休市致股票财务空走ETF兜底,策略数值无意义但链路100%通;周一download恢复走真实选股。
This commit is contained in:
@@ -163,15 +163,11 @@ class SanguoMiniQmtProvider(MiniQMTProvider): # type: ignore[misc]
|
||||
qmt_stocks = [self._normalize_security_code(s) for s in stocks]
|
||||
try:
|
||||
if self.auto_download:
|
||||
try:
|
||||
# 实证 signature: download_financial_data(stock_list, table_list=[])
|
||||
# trading hours Income/CashFlow 常超时,主拿 PershareIndex/Balance/Capital
|
||||
xt.download_financial_data(
|
||||
qmt_stocks,
|
||||
["PershareIndex", "Balance", "Capital"],
|
||||
)
|
||||
except Exception as exc:
|
||||
logger.debug("download_financial_data 失败(继续读缓存): %s", exc)
|
||||
# download_financial_data 阻塞无超时,休市/服务不响应会卡死整个回测;
|
||||
# 包 threading 超时,超时则跳过读缓存(数据可能不全但回测不卡死)
|
||||
self._download_financial_safe(
|
||||
qmt_stocks, ["PershareIndex", "Balance", "Capital"]
|
||||
)
|
||||
fin_data = xt.get_financial_data(qmt_stocks)
|
||||
except Exception as exc:
|
||||
logger.warning("get_financial_data 失败,返空表: %s", exc)
|
||||
@@ -193,6 +189,35 @@ class SanguoMiniQmtProvider(MiniQMTProvider): # type: ignore[misc]
|
||||
df = df.set_index("code", drop=False)
|
||||
return df
|
||||
|
||||
def _download_financial_safe(
|
||||
self, qmt_stocks: List[str], tables: List[str], timeout: float = 120.0
|
||||
) -> None:
|
||||
"""download_financial_data 包 threading 超时。
|
||||
|
||||
xtdata.download_financial_data 阻塞且无超时参数,休市/服务不响应时卡死
|
||||
整个回测。用线程 + join(timeout):超时则放弃(读缓存),daemon 线程随进程退出清理。
|
||||
"""
|
||||
import threading
|
||||
xt = self._ensure_xtdata()
|
||||
err: List[str] = []
|
||||
|
||||
def _worker() -> None:
|
||||
try:
|
||||
xt.download_financial_data(qmt_stocks, tables)
|
||||
except Exception as exc:
|
||||
err.append(str(exc))
|
||||
|
||||
t = threading.Thread(target=_worker, daemon=True)
|
||||
t.start()
|
||||
t.join(timeout)
|
||||
if t.is_alive():
|
||||
logger.warning(
|
||||
"download_financial_data 超时 %.0fs (%d 只),跳过读缓存",
|
||||
timeout, len(qmt_stocks),
|
||||
)
|
||||
elif err:
|
||||
logger.debug("download_financial_data 失败(继续读缓存): %s", err[0])
|
||||
|
||||
# ------------------------ 内部组装 ------------------------
|
||||
def _build_row(
|
||||
self,
|
||||
|
||||
Reference in New Issue
Block a user