aeroxw/tick-stock-panel
1
0
Fork 0
mirror of https://ghfast.top/https://github.com/aeroxw/tick-stock-panel.git synced 2026-09-12 17:54:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix: 修复 6 个 issue (#225/#226/#201/#188/#200/#196)

Browse Source 
后端:
- #225 自定义源分钟K字符串 datetime 不再被 cast 成 null:
  _normalize_minute 对 Utf8 列按常见格式链式解析 (参照 kline_sync 口径)
- #226 自定义源日K/除权因子单批失败只隔离该批 (重试 1 次 + 跳过 +
  warning 汇总), 不再丢弃已成功批次的全部进度
- #201 旧信号回测 _load_panel 加指标 warmup 窗口 (120 交易日保守日历日),
  计算后裁回 [start,end]; 数据不足时自然退化

前端:
- #188 因子回测单标的不再整面板空白: 外层条件改 !error, IC 卡片
  单独守卫并给出需 >=2 只的提示
- #200 因子回测支持调仓频率 (日/周/月) 与滑点 (bp) 配置
- #196 自选页板块筛选新增 ETF 分类, 旧偏好加载时补 ETF 键保持默认可见
This commit is contained in:
shy3130
2026-09-03 13:20:59 +08:00
parent e89ea9becf
commit ef59df4a67
6 changed files with 362 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/data_providers/custom/provider.py
+87 -4
View File
@@ -3,6 +3,7 @@ from __future__ import annotations
import logging
import os
import time
from collections.abc import Callable
from datetime import datetime, timedelta
from pathlib import Path
@@ -132,6 +133,23 @@ class GenericHTTPProvider:
)
return errors
def _request_rows_retry(
self, cfg, symbols: list[str], *, start_time=None, end_time=None, retries: int = 1
) -> list[dict]:
"""单批请求 + 短退避重试。仍失败抛出, 由调用方决定隔离粒度 (#226)。"""
last: Exception | None = None
for attempt in range(retries + 1):
try:
return self._request_rows(
cfg, symbols=symbols, start_time=start_time, end_time=end_time
)
except Exception as e: # noqa: BLE001
last = e
if attempt < retries:
time.sleep(1.0 * (attempt + 1))
assert last is not None
raise last
def get_daily(
self,
symbols: list[str],
@@ -143,15 +161,35 @@ class GenericHTTPProvider:
cfg = self._dataset("daily")
frames: list[pl.DataFrame] = []
chunks = chunked(symbols, cfg.batch)
failed: list[str] = []
for i, chunk in enumerate(chunks):
sleep_between_batches(i, cfg.rpm)
rows = self._request_rows(cfg, symbols=chunk, start_time=start_time, end_time=end_time)
try:
rows = self._request_rows_retry(
cfg, chunk, start_time=start_time, end_time=end_time
)
except Exception as e: # noqa: BLE001
# 单批失败只隔离该批 (#226): 之前任一批 502 会让整个 stage
# 抛异常, 已成功批次的结果留在内存里全部丢弃
failed.extend(chunk)
logger.warning(
"custom daily: batch %d/%d failed (%d symbols), skipped: %s",
i + 1, len(chunks), len(chunk), e,
)
if on_chunk_done:
on_chunk_done(i + 1, len(chunks))
continue
df = self._mapped_frame(cfg, rows)
df = normalize_daily(df, source=self.name)
if not df.is_empty():
frames.append(df)
if on_chunk_done:
on_chunk_done(i + 1, len(chunks))
if failed:
logger.warning(
"custom daily: %d/%d symbols missing due to batch failures: %s",
len(failed), len(symbols), ", ".join(failed[:20]),
)
return pl.concat(frames, how="diagonal_relaxed") if frames else pl.DataFrame()
def get_adj_factors(
@@ -165,15 +203,33 @@ class GenericHTTPProvider:
cfg = self._dataset("adj_factor")
frames: list[pl.DataFrame] = []
chunks = chunked(symbols, cfg.batch)
failed: list[str] = []
for i, chunk in enumerate(chunks):
sleep_between_batches(i, cfg.rpm)
rows = self._request_rows(cfg, symbols=chunk, start_time=start_time, end_time=end_time)
try:
rows = self._request_rows_retry(
cfg, chunk, start_time=start_time, end_time=end_time
)
except Exception as e: # noqa: BLE001
failed.extend(chunk)
logger.warning(
"custom adj_factor: batch %d/%d failed (%d symbols), skipped: %s",
i + 1, len(chunks), len(chunk), e,
)
if on_chunk_done:
on_chunk_done(i + 1, len(chunks))
continue
df = self._mapped_frame(cfg, rows)
df = normalize_adj_factors(df, source=self.name)
if not df.is_empty():
frames.append(df)
if on_chunk_done:
on_chunk_done(i + 1, len(chunks))
if failed:
logger.warning(
"custom adj_factor: %d/%d symbols missing due to batch failures: %s",
len(failed), len(symbols), ", ".join(failed[:20]),
)
return pl.concat(frames, how="diagonal_relaxed") if frames else pl.DataFrame()
def get_realtime(self) -> list[dict]:
@@ -293,12 +349,18 @@ class GenericHTTPProvider:
return pl.DataFrame()
return pl.concat(frames, how="diagonal_relaxed")
@staticmethod
def _normalize_minute(df: pl.DataFrame) -> pl.DataFrame:
@classmethod
def _normalize_minute(cls, df: pl.DataFrame) -> pl.DataFrame:
"""把映射后的 df 规范成 minute canonical 列。"""
if df.is_empty():
return df
if "datetime" in df.columns and df.schema["datetime"] != pl.Datetime("us"):
if df.schema["datetime"] == pl.Utf8:
# 字符串 datetime 直接 cast 会整体置 null (polars 不做字符串解析);
# 先解析再对齐微秒精度 (#225, 参照
# kline_sync._enforce_minute_beijing_wallclock 的处理)。
# Series 级立即解析: 表达式错误要到 collect 才抛, 无法按格式回退
df = df.with_columns(cls._parse_datetime_series(df["datetime"]))
df = df.with_columns(pl.col("datetime").cast(pl.Datetime("us"), strict=False))
for col in ("open", "high", "low", "close", "volume", "amount"):
if col in df.columns:
@@ -306,6 +368,27 @@ class GenericHTTPProvider:
keep = [c for c in ("symbol", "datetime", "open", "high", "low", "close", "volume", "amount") if c in df.columns]
return df.select(keep) if keep else pl.DataFrame()
_DATETIME_STR_FORMATS = (
None, # 自动推断
"%Y-%m-%d %H:%M:%S",
"%Y-%m-%dT%H:%M:%S",
"%Y/%m/%d %H:%M:%S",
"%Y-%m-%d %H:%M",
)
@classmethod
def _parse_datetime_series(cls, s: pl.Series) -> pl.Series:
"""逐格式尝试解析字符串 datetime; 均失败返回全 null (宽松语义)。"""
for fmt in cls._DATETIME_STR_FORMATS:
try:
return (
s.str.to_datetime(strict=False, format=fmt)
if fmt else s.str.to_datetime(strict=False)
)
except Exception: # noqa: BLE001 — 该格式不适用, 换下一个
continue
return pl.Series("datetime", [None] * s.len(), dtype=pl.Datetime("us"))
def test_dataset(self, dataset: str, symbols: list[str] | None = None) -> dict:
cfg = self._dataset(dataset)
test_symbols = symbols or ["000001.SZ"]
backend/app/services/backtest.py
+13 -2
View File
@@ -7,7 +7,7 @@ from __future__ import annotations
import logging
import uuid
from dataclasses import dataclass, field
from datetime import date
from datetime import date, timedelta
from typing import Literal
import numpy as np
@@ -20,6 +20,10 @@ from app.tickflow.repository import KlineRepository
logger = logging.getLogger(__name__)
# 旧信号回测的指标 warmup 日历窗口 (#201): 与 backtest.factor.FACTOR_WARMUP_DAYS
# 同源 (120 交易日 → 保守取日历日), 覆盖 MA60/MACD/BOLL 等最长回看
_WARMUP_CALENDAR_DAYS = 120 * 1.6
# vectorbt 是 optional extras(见 pyproject.toml).未装时只有 backtest 不可用,其他功能正常.
_vbt = None
_vbt_unavailable_reason: str | None = None
@@ -162,11 +166,17 @@ class BacktestService:
try:
from app.tickflow.repository import enriched_dirname
enriched_glob = str(self.repo.store.data_dir / enriched_dirname(asset_type) / "**" / "*.parquet")
# 指标 warmup (#201): MA/MACD/RSI/BOLL 需要区间前的历史窗口,
# 直接按 [start,end] 过滤后 compute_all 会让区间头部的指标失真。
# 与挖掘侧同款公式 (mining_runtime: warmup = max(120, bars*1.6)),
# 此处指标最长回看约 120 交易日, 取保守日历日窗口; 数据不足时
# 自然退化 (有多少算多少)。计算完成后裁回 [start,end]。
warmup_start = start - timedelta(days=_WARMUP_CALENDAR_DAYS)
df = (
scan_enriched_parquet(enriched_glob)
.filter(
(pl.col("symbol").is_in(symbols))
& (pl.col("date") >= start)
& (pl.col("date") >= warmup_start)
& (pl.col("date") <= end)
)
.sort(["date", "symbol"])
@@ -182,6 +192,7 @@ class BacktestService:
# 即时计算指标 + 信号
from app.indicators.pipeline import compute_all
df = compute_all(df)
df = df.filter(pl.col("date") >= start)
# 选择需要的列
needed_cols = [
backend/tests/test_backtest_warmup.py
+66
View File
@@ -0,0 +1,66 @@
"""#201 回归: 旧信号回测的 _load_panel 必须带指标 warmup 窗口。
直接按 [start,end] 过滤后 compute_all, 区间头部的 MA/MACD/RSI 会因缺
历史窗口而失真 (回测起始段信号不可信)。
"""
from __future__ import annotations
from datetime import date, timedelta
from unittest.mock import MagicMock
import polars as pl
from app.services.backtest import BacktestService
def _synthetic_enriched(n_days: int) -> pl.DataFrame:
base = date(2026, 1, 1)
days = [base + timedelta(days=i) for i in range(n_days)]
n = len(days)
closes = [10.0 + (i % 7) * 0.3 + i * 0.01 for i in range(n)]
return pl.DataFrame(
{
"symbol": ["600000.SH"] * n,
"date": days,
"open": [c - 0.05 for c in closes],
"high": [c + 0.1 for c in closes],
"low": [c - 0.1 for c in closes],
"close": closes,
"volume": [10000.0] * n,
"amount": [c * 10000.0 for c in closes],
"raw_close": closes,
"raw_high": [c + 0.1 for c in closes],
"raw_low": [c - 0.1 for c in closes],
}
)
def test_load_panel_warms_up_indicators(monkeypatch) -> None:
df = _synthetic_enriched(250)
monkeypatch.setattr(
"app.services.backtest.scan_enriched_parquet", lambda glob: df.lazy()
)
svc = BacktestService(repo=MagicMock())
start = df["date"][-30]
end = df["date"][-1]
panel = svc._load_panel(["600000.SH"], start, end)
# warmup 行不进入结果面板 (pandas datetime64 与 date 直接比较会类型不符)
assert str(panel["date"].min())[:10] == start.isoformat()
assert str(panel["date"].max())[:10] == end.isoformat()
# 区间首日的指标已有历史窗口可用, 不再是 NaN
first = panel.iloc[0]
assert first["rsi_14"] == first["rsi_14"] # NaN != NaN
def test_load_panel_insufficient_history_degrades_gracefully(monkeypatch) -> None:
# 数据起点晚于 warmup 起点时自然退化: 有多少算多少, 不抛异常
df = _synthetic_enriched(20)
monkeypatch.setattr(
"app.services.backtest.scan_enriched_parquet", lambda glob: df.lazy()
)
svc = BacktestService(repo=MagicMock())
panel = svc._load_panel(["600000.SH"], df["date"][0], df["date"][-1])
assert len(panel) == 20
backend/tests/test_custom_batch_isolation.py
+140
View File
@@ -0,0 +1,140 @@
"""#225/#226 回归: 自定义源分钟K字符串日期解析 + 日K分批失败隔离。"""
from __future__ import annotations
from datetime import date, datetime
import polars as pl
from app.data_providers.custom.config import CustomSourceConfig, DatasetConfig
from app.data_providers.custom.loader import GenericHTTPProvider
def _provider(datasets: dict[str, DatasetConfig]) -> GenericHTTPProvider:
return GenericHTTPProvider(CustomSourceConfig(
name="test_source",
display_name="Test Source",
datasets=datasets,
))
def _daily_config(batch: int = 2) -> DatasetConfig:
return DatasetConfig(
url="https://example.test/daily",
field_map={
"symbol": "symbol", "date": "date", "open": "open", "high": "high",
"low": "low", "close": "close", "volume": "volume", "amount": "amount",
},
batch=batch,
)
# ── #225: 分钟K字符串 datetime 不得被 cast 成 null ────────────────
def test_normalize_minute_parses_string_datetime() -> None:
df = pl.DataFrame(
{
"symbol": ["600000.SH"] * 2,
# 上游 YAML 映射后仍是字符串; 旧代码直接 cast → 全 null (#225)
"datetime": ["2026-09-01 09:35:00", "2026-09-01 09:40:00"],
"close": [10.0, 10.5],
}
)
out = GenericHTTPProvider._normalize_minute(df)
assert out.schema["datetime"] == pl.Datetime("us")
assert out["datetime"].null_count() == 0
assert out["datetime"][0] == datetime(2026, 9, 1, 9, 35)
# 非法字符串维持 strict=False 宽松行为 (null, 不抛异常)
bad = pl.DataFrame(
{"symbol": ["600000.SH"], "datetime": ["not-a-date"], "close": [1.0]}
)
out_bad = GenericHTTPProvider._normalize_minute(bad)
assert out_bad["datetime"].null_count() == 1
def test_normalize_minute_datetime_already_typed_unchanged() -> None:
df = pl.DataFrame(
{
"symbol": ["600000.SH"],
"datetime": [datetime(2026, 9, 1, 9, 35)],
"close": [10.0],
}
)
out = GenericHTTPProvider._normalize_minute(df)
assert out["datetime"][0] == datetime(2026, 9, 1, 9, 35)
# ── #226: get_daily 单批失败只隔离该批 ────────────────────────
def _canonical_frame(symbols: list[str], day: str) -> pl.DataFrame:
return pl.DataFrame(
{
"symbol": symbols,
"date": [date.fromisoformat(day)] * len(symbols),
"open": [10.0] * len(symbols),
"high": [11.0] * len(symbols),
"low": [9.0] * len(symbols),
"close": [10.5] * len(symbols),
"volume": [100.0] * len(symbols),
"amount": [1050.0] * len(symbols),
}
)
def test_get_daily_isolates_failed_batch() -> None:
provider = _provider({"daily": _daily_config(batch=2)})
calls: list[list[str]] = []
def request_rows(cfg, symbols=None, **kwargs):
calls.append(list(symbols))
if symbols == ["s3", "s4"]:
raise RuntimeError("502 Bad Gateway")
return [{"_rows": list(symbols)}]
provider._request_rows = request_rows
provider._mapped_frame = lambda cfg, rows: _canonical_frame(
rows[0]["_rows"], "2026-09-01"
)
df = provider.get_daily(
["s1", "s2", "s3", "s4", "s5", "s6"],
datetime(2026, 8, 1), datetime(2026, 9, 1),
)
# 3 批都请求过 (失败批重试 1 次后跳过、流程继续), 返回第 1、3 批共 4 行
assert calls == [["s1", "s2"], ["s3", "s4"], ["s3", "s4"], ["s5", "s6"]]
assert df.height == 4
assert set(df["symbol"]) == {"s1", "s2", "s5", "s6"}
def test_get_daily_progress_callback_fires_for_failed_batch() -> None:
provider = _provider({"daily": _daily_config(batch=2)})
def request_rows(cfg, symbols=None, **kwargs):
if symbols == ["s3", "s4"]:
raise RuntimeError("timeout")
return [{"_rows": list(symbols)}]
progress: list[tuple[int, int]] = []
provider._request_rows = request_rows
provider._mapped_frame = lambda cfg, rows: _canonical_frame(
rows[0]["_rows"], "2026-09-01"
)
provider.get_daily(
["s1", "s2", "s3", "s4"], datetime(2026, 8, 1), datetime(2026, 9, 1),
on_chunk_done=lambda cur, tot: progress.append((cur, tot)),
)
# 失败批也推进进度, 前端进度条不会卡死
assert progress == [(1, 2), (2, 2)]
def test_get_daily_all_batches_fail_returns_empty() -> None:
provider = _provider({"daily": _daily_config(batch=2)})
def request_rows(cfg, symbols=None, **kwargs):
raise RuntimeError("down")
provider._request_rows = request_rows
df = provider.get_daily(["s1", "s2"], datetime(2026, 8, 1), datetime(2026, 9, 1))
assert df.is_empty()