fanhuadesenlinnn/daily_stock_analysis
1
0
Fork 0
mirror of https://github.com/ZhuLinsen/daily_stock_analysis synced 2026-09-20 10:53:33 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix: preserve alert lifecycle across rolling upgrades

Browse Source
This commit is contained in:
ZhuLinsen
2026-08-29 22:08:43 +08:00
parent 9bddd3b8a7
commit 0acba5d2c3
8 changed files with 105 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
apps/dsa-web/src/components/alerts/AlertMonitorSummary.tsx
View File

@@ -31,7 +31,7 @@ export const AlertMonitorSummary: React.FC<Props> = ({ summary, loading = false
<InlineAlert
variant="warning"
title="存在无法完整归因的触发记录"
message={`无 rule_id：${summary.unattributedTriggerCount}；规则已删除：${summary.orphanedTriggerCount}`}
message={`无 rule_id：${summary.unattributedTriggerCount}；生命周期无法匹配：${summary.orphanedTriggerCount}`}
/>
) : null}
<div className="grid gap-3 md:grid-cols-3">
@@ -76,7 +76,7 @@ export const AlertMonitorSummary: React.FC<Props> = ({ summary, loading = false
</section>
</div>
<div className="flex items-center gap-2 text-xs text-muted-text">
<Link2Off className="h-3.5 w-3.5" />触发归因只按 rule_id 关联，不按股票 target 猜测规则。
<Link2Off className="h-3.5 w-3.5" />触发归因按 rule_id + lifecycle 关联，不按股票 target 猜测规则。
</div>
</div>
) : null}

4
apps/dsa-web/src/components/alerts/__tests__/AlertMonitorSummary.test.tsx
View File

@@ -37,9 +37,9 @@ describe('AlertMonitorSummary', () => {
expect(screen.getByText('监控概览')).toBeInTheDocument();
expect(screen.getByText('25')).toBeInTheDocument();
expect(screen.getByText('41')).toBeInTheDocument();
expect(screen.getByText('无 rule_id：1；规则已删除：2')).toBeInTheDocument();
expect(screen.getByText('无 rule_id：1；生命周期无法匹配：2')).toBeInTheDocument();
expect(screen.getByText('#7 组合回撤')).toBeInTheDocument();
expect(screen.getByText('9 次')).toBeInTheDocument();
expect(screen.getByText(/只按 rule_id 关联/)).toBeInTheDocument();
expect(screen.getByText(/按 rule_id \+ lifecycle 关联/)).toBeInTheDocument();
});
});

2
docs/alerts.md
View File

@@ -440,7 +440,7 @@ worker 会把 `triggered`、`skipped`、`degraded`、`failed` 写入 `alert_trig
- 触发归属使用 `rule_id + rule_lifecycle_id`：每次创建规则都会生成不可复用的 lifecycle ID，worker 在加载规则时固定该 ID，并随触发记录一起写入。即使旧 worker 已经开始计算、期间规则被删除且 SQLite 复用了整数 ID，迟到的旧触发也不会归到新规则。`rule_id` 为空的记录计入 `unattributed_trigger_count`；引用已不存在规则或 lifecycle 不匹配的记录计入 `orphaned_trigger_count`。
- 摘要的总数、状态分组、规则分组和孤儿计数在同一个 SQLite 读事务快照内完成，后台 worker 同时写入触发记录时不会返回“总数小于状态分组之和”的自相矛盾结果。
该概览是 Issue #2281 的基础阶段，只解决全局统计与可靠归属。它不引入“论文失效”规则类型，也不把普通告警命名为失效信号；`impact`、`affected_entities` 和新的 thesis invalidation 规则语义留待后续独立 PR。启动时会为既有 SQLite 的 `alert_rules` / `alert_triggers` 增加 lifecycle 列；既有规则会获得 lifecycle，但迁移前的触发记录无法证明属于哪一代规则，因此不按时间戳猜测回填，统一保留为 orphan。多实例同时首次升级时，重复加列会安全收敛。回滚代码不会自动删除新增列或历史数据，旧版本会忽略这些附加列。
该概览是 Issue #2281 的基础阶段，只解决全局统计与可靠归属。它不引入“论文失效”规则类型，也不把普通告警命名为失效信号；`impact`、`affected_entities` 和新的 thesis invalidation 规则语义留待后续独立 PR。启动时会为既有 SQLite 的 `alert_rules` / `alert_triggers` 增加可空 lifecycle 列；既有规则会获得 lifecycle，但迁移前的触发记录无法证明属于哪一代规则，因此不按时间戳猜测回填，统一保留为“生命周期无法匹配”。多实例同时首次升级时，重复加列会安全收敛；滚动升级期间旧进程创建的 lifecycle-less 规则，会在新 worker 评估前原子补齐。回滚代码不会自动删除新增列或历史数据，但旧版本仍可向可空列插入并会忽略附加列。
## Phase 边界

27
src/repositories/alert_repo.py
View File

@@ -8,8 +8,9 @@ from __future__ import annotations
from datetime import datetime
from typing import Any, Dict, List, Optional, Tuple
import uuid
from sqlalchemy import and_, case, delete, desc, func, select
from sqlalchemy import and_, case, delete, desc, func, or_, select, update
from src.storage import (
AlertCooldownRecord,
@@ -40,6 +41,30 @@ class AlertRepository:
select(AlertRuleRecord).where(AlertRuleRecord.id == rule_id).limit(1)
).scalar_one_or_none()
def ensure_rule_lifecycle(self, rule_id: int) -> Optional[str]:
"""Atomically assign a lifecycle to a rule created by an older process."""
lifecycle_id = str(uuid.uuid4())
def assign(session) -> Optional[str]:
session.execute(
update(AlertRuleRecord)
.where(
AlertRuleRecord.id == rule_id,
or_(
AlertRuleRecord.lifecycle_id.is_(None),
AlertRuleRecord.lifecycle_id == "",
),
)
.values(lifecycle_id=lifecycle_id)
)
return session.execute(
select(AlertRuleRecord.lifecycle_id)
.where(AlertRuleRecord.id == rule_id)
.limit(1)
).scalar_one_or_none()
return self.db._run_write_transaction("ensure_alert_rule_lifecycle", assign)
def update_rule(self, rule_id: int, fields: Dict[str, Any]) -> Optional[AlertRuleRecord]:
with self.db.get_session() as session:
row = session.execute(

7
src/services/alert_worker.py
View File

@@ -207,6 +207,11 @@ class AlertWorker:
for row in self.service.repo.list_enabled_rules(limit=ALERT_WORKER_RULE_LIMIT):
try:
lifecycle_id = getattr(row, "lifecycle_id", None)
if not lifecycle_id:
lifecycle_id = self.service.repo.ensure_rule_lifecycle(int(row.id))
if not lifecycle_id:
raise ValueError("persisted alert rule has no lifecycle identity")
cooldown_policy = self.service._load_json(row.cooldown_policy, default=None)
for payload in self.service.build_runtime_payloads(row, config=config, include_overflow_payload=False):
if len(runtime_rules) >= ALERT_WORKER_RULE_LIMIT:
@@ -224,7 +229,7 @@ class AlertWorker:
cooldown_policy=cooldown_policy,
effective_target=payload.effective_target,
display_target=payload.display_target,
rule_lifecycle_id=str(row.lifecycle_id),
rule_lifecycle_id=str(lifecycle_id),
)
)
seen_keys.add(payload.key)

2
src/storage.py
View File

@@ -938,7 +938,7 @@ class AlertRuleRecord(Base):
__tablename__ = 'alert_rules'
id = Column(Integer, primary_key=True, autoincrement=True)
lifecycle_id = Column(String(36), nullable=False, default=lambda: str(uuid.uuid4()), index=True)
lifecycle_id = Column(String(36), nullable=True, default=lambda: str(uuid.uuid4()), index=True)
name = Column(String(64), nullable=False)
target_scope = Column(String(32), nullable=False, default='single_symbol', index=True)
target = Column(String(64), nullable=False, index=True)

32
tests/test_alert_api.py
View File

@@ -951,6 +951,38 @@ class AlertApiTestCase(unittest.TestCase):
self.assertEqual(by_rule_id[rule["id"]]["trigger_count"], 0)
self.assertEqual(payload["orphaned_trigger_count"], 1)
def test_rolling_upgrade_assigns_lifecycle_to_rule_created_by_old_process(self) -> None:
connection = sqlite3.connect(self.db_path)
try:
cursor = connection.execute(
"INSERT INTO alert_rules "
"(name, target_scope, target, alert_type, parameters, severity, enabled, source) "
"VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
(
"legacy-process-rule",
"single_symbol",
"600519",
"price_cross",
'{"direction":"above","price":1800}',
"warning",
True,
"api",
),
)
rule_id = int(cursor.lastrowid)
connection.commit()
finally:
connection.close()
repo = AlertRepository(self.db)
self.assertIsNone(repo.get_rule(rule_id).lifecycle_id)
assigned = repo.ensure_rule_lifecycle(rule_id)
self.assertIsNotNone(assigned)
self.assertEqual(repo.get_rule(rule_id).lifecycle_id, assigned)
self.assertEqual(repo.ensure_rule_lifecycle(rule_id), assigned)
def test_monitor_summary_uses_one_read_snapshot_during_concurrent_trigger_write(self) -> None:
rule = self._create_rule({"name": "snapshot", "target": "600519"})
AlertRepository(self.db).create_trigger(

35
tests/test_alert_worker.py
View File

@@ -5,6 +5,7 @@ from __future__ import annotations
import json
import os
import sqlite3
import tempfile
import unittest
from datetime import date
@@ -316,6 +317,40 @@ class AlertWorkerTestCase(unittest.TestCase):
notifier.send_with_results.side_effect = list(results)
return notifier
def test_load_runtime_rules_repairs_rule_created_by_old_process(self) -> None:
connection = sqlite3.connect(self.db_path)
try:
cursor = connection.execute(
"INSERT INTO alert_rules "
"(name, target_scope, target, alert_type, parameters, severity, enabled, source) "
"VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
(
"legacy-process-rule",
"single_symbol",
"600519",
"price_cross",
'{"direction":"above","price":1800}',
"warning",
True,
"api",
),
)
rule_id = int(cursor.lastrowid)
connection.commit()
finally:
connection.close()
worker = AlertWorker(config_provider=lambda: self._config(), service=self.service)
runtime_rules = worker._load_runtime_rules(self._config())
self.assertEqual(len(runtime_rules), 1)
self.assertEqual(runtime_rules[0].source, "db")
self.assertIsNotNone(runtime_rules[0].rule_lifecycle_id)
self.assertEqual(
self.service.repo.get_rule(rule_id).lifecycle_id,
runtime_rules[0].rule_lifecycle_id,
)
def test_p6_triggered_stock_alert_links_latest_active_decision_signal(self) -> None:
self._create_rule(target="600519")
signal_service = DecisionSignalService()