leefer/xiaobaifupan
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: track screener candidates across five trading days

Browse Source
This commit is contained in:
leefer
2026-07-22 23:55:33 +08:00
parent 50dbb8a3ff
commit 5ae2791dd3
9 changed files with 510 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api_access.py
+2
View File
@@ -9,6 +9,7 @@ AccessRole = Literal["authenticated", "member", "admin"]
MEMBER_GET_PATHS = frozenset(
{
"/api/screener/setup",
"/api/screener/tracking",
"/api/mentors/setup",
"/api/mentors/messages",
"/api/heaven/setup",
@@ -21,6 +22,7 @@ MEMBER_POST_PATHS = frozenset(
"/api/screener/compile",
"/api/screener/strategies",
"/api/screener/run",
"/api/screener/tracking/refresh",
"/api/mentors/chat",
"/api/heaven/hexagram",
"/api/heaven/personal",
database.py
+105
View File
@@ -256,6 +256,27 @@ class ReviewDatabase:
CREATE INDEX IF NOT EXISTS idx_mentor_messages_conversation
ON mentor_messages(user_id, mentor_id, trade_date, id DESC);
CREATE TABLE IF NOT EXISTS strategy_tracks (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL,
run_id INTEGER NOT NULL,
selection_date TEXT NOT NULL,
strategy_name TEXT NOT NULL,
ts_code TEXT NOT NULL,
code TEXT NOT NULL,
name TEXT NOT NULL,
sector TEXT NOT NULL DEFAULT '',
entry_price REAL NOT NULL,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
UNIQUE(user_id, run_id, ts_code),
FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE,
FOREIGN KEY (run_id) REFERENCES screener_runs(id) ON DELETE CASCADE
);
CREATE INDEX IF NOT EXISTS idx_strategy_tracks_user_run
ON strategy_tracks(user_id, run_id DESC, id);
"""
)
user_columns = {
@@ -1321,6 +1342,90 @@ class ReviewDatabase:
)
return int(cursor.rowcount)
def save_strategy_tracks(
self,
user_id: int,
run_id: int,
selection_date: str,
strategy_name: str,
candidates: list[dict[str, Any]],
) -> int:
now = datetime.now().astimezone().isoformat(timespec="seconds")
values = []
for item in candidates:
ts_code = str(item.get("ts_code") or "").strip()
code = str(item.get("code") or ts_code.split(".")[0]).strip()
entry_price = float(item.get("price") or 0)
if not ts_code or not code or entry_price <= 0:
continue
values.append(
(
int(user_id), int(run_id), selection_date, strategy_name, ts_code, code,
str(item.get("name") or "--"), str(item.get("sector") or "其他"),
entry_price, now, now,
)
)
with self.connect() as connection:
connection.executemany(
"""
INSERT INTO strategy_tracks
(user_id, run_id, selection_date, strategy_name, ts_code, code,
name, sector, entry_price, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(user_id, run_id, ts_code) DO UPDATE SET
name=excluded.name, sector=excluded.sector,
entry_price=excluded.entry_price, updated_at=excluded.updated_at
""",
values,
)
return len(values)
def list_strategy_tracks(self, user_id: int, limit_batches: int = 12) -> list[dict[str, Any]]:
limit_batches = max(1, min(50, int(limit_batches)))
with self.connect() as connection:
rows = connection.execute(
"""
SELECT * FROM strategy_tracks
WHERE user_id = ? AND run_id IN (
SELECT run_id FROM strategy_tracks WHERE user_id = ?
GROUP BY run_id ORDER BY run_id DESC LIMIT ?
)
ORDER BY run_id DESC, id
""",
(int(user_id), int(user_id), limit_batches),
).fetchall()
return [dict(row) for row in rows]
def load_tracking_bars(
self, targets: list[tuple[str, str]], limit: int = 5
) -> dict[tuple[str, str], list[dict[str, Any]]]:
unique_targets = set(targets)
if not unique_targets:
return {}
codes = sorted({ts_code for ts_code, _ in unique_targets})
earliest_date = min(selection_date for _, selection_date in unique_targets)
placeholders = ",".join("?" for _ in codes)
with self.connect() as connection:
rows = connection.execute(
f"""
SELECT ts_code, trade_date, open, high, low, close FROM daily_bars
WHERE ts_code IN ({placeholders}) AND trade_date > ?
ORDER BY ts_code, trade_date
""",
[*codes, earliest_date],
).fetchall()
by_code: dict[str, list[dict[str, Any]]] = {}
for row in rows:
item = dict(row)
by_code.setdefault(str(item["ts_code"]), []).append(item)
row_limit = max(1, min(20, int(limit)))
return {
(ts_code, selection_date): [
row for row in by_code.get(ts_code, []) if row["trade_date"] > selection_date
][:row_limit]
for ts_code, selection_date in unique_targets
}
def start_sync(self, trade_date: str, source: str) -> int:
started_at = datetime.now().astimezone().isoformat(timespec="seconds")
with self.connect() as connection:
server.py
+53 -1
View File
@@ -62,6 +62,7 @@ from sentiment_engine import (
build_sentiment_history,
latest_contiguous_history,
)
from strategy_tracking import StrategyTrackingService
from tushare_client import TushareClient, TushareError
@@ -127,6 +128,7 @@ class DashboardService:
self._request_context = threading.local()
self._system_credentials = self._load_system_credentials(environment_credentials)
self.screener = ScreenerEngine(self.database)
self.strategy_tracking = StrategyTrackingService(self.database)
self.mentor_skills = MentorSkillRegistry(MENTOR_SKILLS_DIR)
self.realtime_aggregator = WebRealtimeAggregator()
self.screener.ensure_builtin_strategies()
@@ -1082,6 +1084,26 @@ class DashboardService:
"latest_result": self.database.latest_screener_run(self.current_user_id, normalized_date),
}
def screener_tracking(self, limit: int = 12) -> dict[str, Any]:
return self.strategy_tracking.list_tracking(self.current_user_id, limit)
def refresh_screener_tracking(self, trade_date: str) -> dict[str, Any]:
normalized_date = normalize_date(trade_date)
notice = ""
if self.configured:
try:
FactorDataService(self.database, TushareClient(self.token)).sync(
normalized_date, 15
)
except TushareError:
notice = "最新日线暂未补齐，已按现有数据更新跟踪。"
else:
notice = "公共行情尚未配置，已按现有数据更新跟踪。"
return {
"tracking": self.screener_tracking(),
"notice": notice,
}
def sync_screener_data(self, trade_date: str, lookback: int = 45) -> dict[str, Any]:
if not self.configured:
raise ValueError("请先配置 Tushare Token。")
@@ -2570,11 +2592,19 @@ class DashboardService:
realtime_snapshot = TushareClient(self.token).realtime_factor_snapshot(trade_date)
except TushareError as exc:
raise ValueError(f"实时选股行情不可用，已停止筛选：{exc}") from exc
return self.screener.screen(
result = self.screener.screen(
self.current_user_id, trade_date, formula, regime, strategy_name,
bool(payload.get("run_backtest", True)),
realtime_snapshot,
)
self.strategy_tracking.record_run(
self.current_user_id,
int(result.get("meta", {}).get("run_id") or 0),
normalize_date(str(result.get("meta", {}).get("trade_date") or trade_date)),
strategy_name,
list(result.get("candidates") or []),
)
return result
def get_dragon_tiger(self, trade_date: str, force: bool = False) -> dict[str, Any]:
normalized_date = normalize_date(trade_date)
@@ -3465,6 +3495,15 @@ class RequestHandler(BaseHTTPRequestHandler):
except ValueError as exc:
self.send_json({"error": str(exc)}, HTTPStatus.BAD_REQUEST)
return
if parsed.path == "/api/screener/tracking":
query = parse_qs(parsed.query)
try:
self.send_json(
SERVICE.screener_tracking(int(query.get("limit", ["12"])[0]))
)
except (TypeError, ValueError) as exc:
self.send_json({"error": str(exc)}, HTTPStatus.BAD_REQUEST)
return
if parsed.path == "/api/mentors/setup":
query = parse_qs(parsed.query)
trade_date = query.get("trade_date", [date.today().isoformat()])[0]
@@ -3577,6 +3616,9 @@ class RequestHandler(BaseHTTPRequestHandler):
if parsed.path == "/api/screener/run":
self.run_screener()
return
if parsed.path == "/api/screener/tracking/refresh":
self.refresh_screener_tracking()
return
if parsed.path == "/api/mentors/chat":
self.mentor_chat()
return
@@ -4010,6 +4052,16 @@ class RequestHandler(BaseHTTPRequestHandler):
except Exception as exc:
self.send_json({"error": f"选股执行失败：{exc}"}, HTTPStatus.INTERNAL_SERVER_ERROR)
def refresh_screener_tracking(self) -> None:
try:
body = self.read_json_body(True)
trade_date = str(body.get("trade_date") or date.today().isoformat())
self.send_json({"ok": True, **SERVICE.refresh_screener_tracking(trade_date)})
except (ValueError, json.JSONDecodeError) as exc:
self.send_json({"error": str(exc)}, HTTPStatus.BAD_REQUEST)
except Exception as exc:
self.send_json({"error": f"跟踪刷新失败：{exc}"}, HTTPStatus.INTERNAL_SERVER_ERROR)
def mentor_chat(self) -> None:
try:
body = self.read_json_body()
static/app.js
+76
View File
@@ -43,6 +43,7 @@ const state = {
selectedRegime: "",
selectedStrategy: null,
screenerResult: null,
screenerTracking: null,
screenerMobileView: "strategy",
sentimentHistory: null,
sentimentRange: 20,
@@ -485,6 +486,7 @@ function bindEvents() {
document.querySelector("#backfillButton").addEventListener("click", backfillData);
document.querySelector("#factorSyncButton").addEventListener("click", syncFactorData);
document.querySelector("#screenerRunButton").addEventListener("click", runScreener);
document.querySelector("#refreshTrackingButton").addEventListener("click", refreshScreenerTracking);
document.querySelectorAll("[data-screener-mobile-view]").forEach((button) => {
button.addEventListener("click", () => selectScreenerMobileView(button.dataset.screenerMobileView));
});
@@ -1590,6 +1592,7 @@ async function loadScreenerSetup() {
|| null;
renderScreenerSetup();
if (state.screenerResult) renderScreenerResult();
await loadScreenerTracking();
} catch (error) {
showToast(error.message || "选股配置加载失败");
}
@@ -1805,6 +1808,7 @@ async function runScreener() {
});
state.screenerResult = payload.result;
renderScreenerResult();
await loadScreenerTracking(true);
setText("screenerRunStatus", `完成 · ${payload.result.candidates.length} 只`);
updateBacktestTaskStatus();
if (window.innerWidth <= 720) selectScreenerMobileView("results");
@@ -3557,6 +3561,78 @@ function renderScreenerResult() {
updateBacktestTaskStatus();
}
async function loadScreenerTracking(force = false) {
if (state.screenerTracking && !force) {
renderScreenerTracking();
return;
}
try {
state.screenerTracking = await apiRequest("/api/screener/tracking?limit=12");
renderScreenerTracking();
} catch (error) {
showToast(error.message || "策略跟踪加载失败");
}
}
async function refreshScreenerTracking() {
const button = document.querySelector("#refreshTrackingButton");
button.disabled = true;
setStatus("正在更新策略跟踪");
try {
const payload = await apiRequest("/api/screener/tracking/refresh", "POST", {
trade_date: elements.tradeDate.value,
});
state.screenerTracking = payload.tracking;
renderScreenerTracking();
if (payload.notice) showToast(payload.notice);
setStatus("策略跟踪已更新");
} catch (error) {
showToast(error.message || "策略跟踪刷新失败");
setStatus("策略跟踪刷新失败");
} finally {
button.disabled = false;
}
}
function renderScreenerTracking() {
const payload = state.screenerTracking || { batches: [], summary: {} };
const batches = payload.batches || [];
const rows = batches.flatMap((batch) => (batch.items || []).map((item) => ({
...item,
selection_date: batch.selection_date,
strategy_name: batch.strategy_name,
})));
setText("trackingBatchCount", `${batches.length} 批`);
const summary = payload.summary || {};
document.querySelector("#trackingSummary").innerHTML = [
["跟踪标的", `${number(summary.total)} 只`],
["已有 T+1", `${number(summary.observed)} 只`],
["T+1 胜率", trackingPercent(summary.t1_win_rate)],
["T+5 胜率", trackingPercent(summary.t5_win_rate)],
["T+5 平均", trackingReturn(summary.average_t5)],
].map(([label, value]) => `<div><span>${label}</span><strong>${value}</strong></div>`).join("");
document.querySelector("#trackingEmpty").hidden = rows.length > 0;
document.querySelector("#trackingTableBody").innerHTML = rows.map((row) => `
<tr data-code="${escapeHtml(row.code)}">
<td>${displayCompactDate(row.selection_date)}</td>
<td class="tracking-strategy" title="${escapeHtml(row.strategy_name)}">${escapeHtml(row.strategy_name)}</td>
<td><span class="stock-cell"><strong>${escapeHtml(row.name)}</strong><small class="stock-code">${escapeHtml(row.code)}</small></span></td>
<td class="number">${formatNumber(row.entry_price, 2)}</td>
${["t1_open", "t1_close", "t3_close", "t5_close", "max_gain", "max_drawdown"].map((key) => `<td class="number ${row[key] == null ? "" : changeClass(row[key])}">${trackingReturn(row[key])}</td>`).join("")}
<td><span class="tracking-status ${row.status === "已完成" ? "complete" : row.observed_days ? "active" : "pending"}">${escapeHtml(row.status)}</span></td>
</tr>
`).join("");
bindStockRows(document.querySelector("#trackingTableBody"));
}
function trackingReturn(value) {
return value == null ? "--" : `${signed(value)}%`;
}
function trackingPercent(value) {
return value == null ? "--" : `${formatNumber(value, 1)}%`;
}
function renderBacktest(backtest) {
const panel = document.querySelector("#backtestPanel");
panel.hidden = !backtest;
static/index.html
+15
View File
@@ -681,6 +681,7 @@
<div class="toolbar-controls">
<button id="factorSyncButton" class="button" type="button">同步因子数据</button>
<button id="screenerRunButton" class="button primary" type="button">执行选股</button>
<button id="refreshTrackingButton" class="button" type="button"><i data-lucide="refresh-cw"></i>刷新跟踪</button>
<button id="screenerExportButton" class="button" type="button">导出 CSV</button>
</div>
</div>
@@ -768,6 +769,20 @@
<div id="screenerEmpty" class="empty-state">尚未执行选股</div>
</div>
</div>
<section class="strategy-tracking-panel" aria-labelledby="strategyTrackingTitle">
<div class="section-toolbar result-toolbar">
<div class="section-title-group"><h2 id="strategyTrackingTitle">策略持续跟踪</h2><span id="trackingBatchCount" class="count-badge">0 批</span></div>
<span class="section-subtitle">以入选价为基准，跟踪后续五个交易日</span>
</div>
<div id="trackingSummary" class="tracking-summary"></div>
<div class="table-frame tracking-table-frame">
<table class="data-table tracking-table">
<thead><tr><th>入选日</th><th>策略</th><th>股票</th><th class="number">入选价</th><th class="number">T+1 开</th><th class="number">T+1 收</th><th class="number">T+3</th><th class="number">T+5</th><th class="number">最大涨幅</th><th class="number">最大回撤</th><th>状态</th></tr></thead>
<tbody id="trackingTableBody"></tbody>
</table>
<div id="trackingEmpty" class="empty-state">执行选股后自动开始跟踪</div>
</div>
</section>
</section>
<section id="mentorView" class="workspace-view member-feature-view">
static/styles.css
+49
View File
@@ -11358,3 +11358,52 @@ button.account-role-badge:focus-visible { outline: 2px solid var(--blue); outlin
transition: none;
}
}
.strategy-tracking-panel {
margin-top: 18px;
border-top: 1px solid var(--border);
background: var(--surface);
}
.tracking-summary {
display: grid;
grid-template-columns: repeat(5, minmax(0, 1fr));
border-top: 1px solid var(--border);
border-bottom: 1px solid var(--border);
background: var(--surface-muted);
}
.tracking-summary > div {
min-width: 0;
padding: 11px 16px;
border-right: 1px solid var(--border);
}
.tracking-summary > div:last-child { border-right: 0; }
.tracking-summary span { display: block; color: var(--text-secondary); font-size: 11px; }
.tracking-summary strong { display: block; margin-top: 4px; font-size: 15px; font-variant-numeric: tabular-nums; }
.tracking-table-frame { border: 0; border-radius: 0; }
.tracking-table { min-width: 1080px; }
.tracking-strategy { max-width: 170px; overflow: hidden; text-overflow: ellipsis; white-space: nowrap; }
.tracking-status {
display: inline-flex;
min-height: 24px;
align-items: center;
padding: 0 8px;
border: 1px solid var(--border);
border-radius: 4px;
color: var(--text-secondary);
font-size: 11px;
white-space: nowrap;
}
.tracking-status.active { border-color: #c5d7ed; background: var(--action-soft); color: var(--action); }
.tracking-status.complete { border-color: #b9dfcf; background: var(--market-down-soft); color: var(--market-down); }
@media (max-width: 720px) {
.tracking-summary { grid-template-columns: repeat(2, minmax(0, 1fr)); }
.tracking-summary > div { border-bottom: 1px solid var(--border); }
.tracking-summary > div:nth-child(2n) { border-right: 0; }
.strategy-tracking-panel .section-toolbar { align-items: flex-start; }
}
strategy_tracking.py
+104
View File
@@ -0,0 +1,104 @@
from __future__ import annotations
from typing import Any
from database import ReviewDatabase
class StrategyTrackingService:
def __init__(self, database: ReviewDatabase) -> None:
self.database = database
def record_run(
self,
user_id: int,
run_id: int,
selection_date: str,
strategy_name: str,
candidates: list[dict[str, Any]],
) -> int:
return self.database.save_strategy_tracks(
user_id, run_id, selection_date, strategy_name, candidates
)
def list_tracking(self, user_id: int, limit_batches: int = 12) -> dict[str, Any]:
tracks = self.database.list_strategy_tracks(user_id, limit_batches)
if not tracks:
return {"batches": [], "summary": self._summary([])}
bars = self.database.load_tracking_bars(
[(item["ts_code"], item["selection_date"]) for item in tracks], 5
)
batches: dict[int, dict[str, Any]] = {}
all_items: list[dict[str, Any]] = []
for track in tracks:
key = (track["ts_code"], track["selection_date"])
metrics = self.calculate_metrics(float(track["entry_price"]), bars.get(key, []))
item = {
"id": track["id"],
"code": track["code"],
"name": track["name"],
"sector": track["sector"],
"entry_price": round(float(track["entry_price"]), 2),
**metrics,
}
all_items.append(item)
batch = batches.setdefault(
int(track["run_id"]),
{
"run_id": int(track["run_id"]),
"selection_date": track["selection_date"],
"strategy_name": track["strategy_name"],
"items": [],
},
)
batch["items"].append(item)
ordered = list(batches.values())
for batch in ordered:
batch["summary"] = self._summary(batch["items"])
return {"batches": ordered, "summary": self._summary(all_items)}
@staticmethod
def calculate_metrics(entry_price: float, bars: list[dict[str, Any]]) -> dict[str, Any]:
valid = [row for row in bars[:5] if float(row.get("close") or 0) > 0]
if entry_price <= 0 or not valid:
return {
"observed_days": 0,
"status": "等待 T+1",
"t1_open": None,
"t1_close": None,
"t3_close": None,
"t5_close": None,
"max_gain": None,
"max_drawdown": None,
}
def change(price: Any) -> float:
return round((float(price or 0) / entry_price - 1) * 100, 2)
observed = len(valid)
return {
"observed_days": observed,
"status": "已完成" if observed >= 5 else f"跟踪中 {observed}/5",
"t1_open": change(valid[0]["open"]),
"t1_close": change(valid[0]["close"]),
"t3_close": change(valid[2]["close"]) if observed >= 3 else None,
"t5_close": change(valid[4]["close"]) if observed >= 5 else None,
"max_gain": max(change(row["high"]) for row in valid),
"max_drawdown": min(change(row["low"]) for row in valid),
}
@staticmethod
def _summary(items: list[dict[str, Any]]) -> dict[str, Any]:
completed = [item for item in items if item.get("t5_close") is not None]
t1 = [float(item["t1_close"]) for item in items if item.get("t1_close") is not None]
t5 = [float(item["t5_close"]) for item in completed]
return {
"total": len(items),
"observed": len(t1),
"completed": len(completed),
"t1_win_rate": round(sum(value > 0 for value in t1) / len(t1) * 100, 1) if t1 else None,
"t5_win_rate": round(sum(value > 0 for value in t5) / len(t5) * 100, 1) if t5 else None,
"average_t5": round(sum(t5) / len(t5), 2) if t5 else None,
}
tests/test_api_access.py
+2
View File
@@ -9,9 +9,11 @@ class ApiAccessPolicyTests(unittest.TestCase):
def test_member_workspaces_are_consistently_protected(self):
cases = {
("GET", "/api/screener/setup"): "member",
("GET", "/api/screener/tracking"): "member",
("GET", "/api/mentors/messages"): "member",
("GET", "/api/heaven/setup"): "member",
("POST", "/api/screener/run"): "member",
("POST", "/api/screener/tracking/refresh"): "member",
("POST", "/api/mentors/chat"): "member",
("POST", "/api/heaven/interpret"): "member",
("DELETE", "/api/screener/strategies/42"): "member",
tests/test_strategy_tracking.py
+104
View File
@@ -0,0 +1,104 @@
from __future__ import annotations
import tempfile
import unittest
from pathlib import Path
from database import ReviewDatabase
from strategy_tracking import StrategyTrackingService
class StrategyTrackingTests(unittest.TestCase):
def setUp(self) -> None:
self.temp = tempfile.TemporaryDirectory()
self.database = ReviewDatabase(Path(self.temp.name) / "review.db")
self.owner = self.database.create_user("track_owner", "salt", "hash")
self.other = self.database.create_user("track_other", "salt", "hash")
self.service = StrategyTrackingService(self.database)
self.run_id = self.database.save_screener_run(
self.owner["id"], "20260710", "repair", "修复策略", {}, {"meta": {}}
)
def tearDown(self) -> None:
self.temp.cleanup()
def test_run_is_recorded_once_and_private_to_owner(self):
candidates = [
{
"ts_code": "002141.SZ",
"code": "002141",
"name": "贤丰控股",
"sector": "油气开采",
"price": 10,
}
]
self.assertEqual(
self.service.record_run(
self.owner["id"], self.run_id, "20260710", "修复策略", candidates
),
1,
)
self.service.record_run(
self.owner["id"], self.run_id, "20260710", "修复策略", candidates
)
self.assertEqual(len(self.database.list_strategy_tracks(self.owner["id"])), 1)
self.assertEqual(self.database.list_strategy_tracks(self.other["id"]), [])
def test_tracking_uses_next_five_trading_bars(self):
self.service.record_run(
self.owner["id"],
self.run_id,
"20260710",
"修复策略",
[{"ts_code": "002141.SZ", "code": "002141", "name": "贤丰控股", "sector": "油气开采", "price": 10}],
)
rows = []
values = (
("20260713", 10.2, 10.8, 9.8, 10.5),
("20260714", 10.5, 11.0, 10.1, 10.8),
("20260715", 10.8, 11.5, 10.4, 11.2),
("20260716", 11.2, 11.4, 9.5, 9.8),
("20260717", 9.8, 10.3, 9.0, 10.0),
("20260720", 10.0, 20.0, 1.0, 19.0),
)
for trade_date, open_price, high, low, close in values:
rows.append(
{
"trade_date": trade_date,
"ts_code": "002141.SZ",
"open": open_price,
"high": high,
"low": low,
"close": close,
}
)
self.database.upsert_daily_bars(rows)
payload = self.service.list_tracking(self.owner["id"])
item = payload["batches"][0]["items"][0]
self.assertEqual(item["status"], "已完成")
self.assertEqual(item["t1_open"], 2.0)
self.assertEqual(item["t1_close"], 5.0)
self.assertEqual(item["t3_close"], 12.0)
self.assertEqual(item["t5_close"], 0.0)
self.assertEqual(item["max_gain"], 15.0)
self.assertEqual(item["max_drawdown"], -10.0)
self.assertEqual(payload["summary"]["t1_win_rate"], 100.0)
self.assertEqual(payload["summary"]["t5_win_rate"], 0.0)
def test_partial_tracking_reports_available_days(self):
metrics = self.service.calculate_metrics(
20,
[
{"open": 20, "high": 21, "low": 19, "close": 20.5},
{"open": 20.5, "high": 22, "low": 20, "close": 21},
],
)
self.assertEqual(metrics["status"], "跟踪中 2/5")
self.assertIsNone(metrics["t3_close"])
self.assertEqual(metrics["max_gain"], 10.0)
self.assertEqual(metrics["max_drawdown"], -5.0)
if __name__ == "__main__":
unittest.main()