rebuild(screener): add controlled formulas and rolling backtests

next/backend/features/screener/backtest.py

@@ -0,0 +1,84 @@
+from __future__ import annotations
+
+from typing import Any
+
+from backend.features.screener.engine import execute_formula
+
+MINIMUM_STABLE_SAMPLES = 20
+FORWARD_TRADING_DAYS = 3
+
+
+def rolling_backtest(
+    snapshots: list[dict[str, Any]], formula: dict[str, Any]
+) -> dict[str, Any]:
+    returns: list[float] = []
+    evaluated_dates: list[str] = []
+    for index in range(max(0, len(snapshots) - FORWARD_TRADING_DAYS)):
+        current = snapshots[index]
+        future = snapshots[index + FORWARD_TRADING_DAYS]
+        outcome = execute_formula(current["rows"], formula, current["coverage"])
+        if outcome["status"] not in {"completed", "no_signal"}:
+            continue
+        evaluated_dates.append(str(current["trade_date"]))
+        future_closes = {
+            str(row.get("identifier")): _positive(row.get("close"))
+            for row in future["rows"]
+        }
+        for candidate in outcome["items"]:
+            entry = _positive(candidate.get("close"))
+            future_close = future_closes.get(str(candidate.get("identifier")))
+            if entry is None or future_close is None:
+                continue
+            returns.append((future_close / entry - 1) * 100)
+    sample_size = len(returns)
+    stable = sample_size >= MINIMUM_STABLE_SAMPLES
+    period_start = evaluated_dates[0] if evaluated_dates else None
+    period_end = evaluated_dates[-1] if evaluated_dates else None
+    return {
+        "sample_size": sample_size,
+        "evaluated_dates": len(evaluated_dates),
+        "stable": stable,
+        "win_rate": (
+            round(sum(value > 0 for value in returns) / sample_size * 100, 1)
+            if stable
+            else None
+        ),
+        "average_return_3d": (
+            round(sum(returns) / sample_size, 2) if stable else None
+        ),
+        "period_start": period_start,
+        "period_end": period_end,
+        "message": (
+            "历史估计已达到最低样本要求"
+            if stable
+            else (
+                f"小样本：当前{sample_size}个有效样本，"
+                f"满{MINIMUM_STABLE_SAMPLES}个后显示胜率与平均收益"
+            )
+        ),
+    }
+
+
+def attach_historical_estimate(
+    run: dict[str, Any], backtest: dict[str, Any] | None
+) -> dict[str, Any]:
+    run["backtest"] = backtest
+    if not backtest or not backtest.get("stable"):
+        return run
+    win_rate = float(backtest["win_rate"])
+    for item in run.get("items") or []:
+        score = item.get("score_display")
+        item["historical_estimate"] = (
+            round(win_rate * 0.65 + float(score) * 0.35, 1)
+            if isinstance(score, (int, float))
+            else None
+        )
+    return run
+
+
+def _positive(value: Any) -> float | None:
+    try:
+        number = float(value)
+    except (TypeError, ValueError):
+        return None
+    return number if number > 0 else None