diff --git a/apps/dsa-web/src/components/alerts/AlertMonitorSummary.tsx b/apps/dsa-web/src/components/alerts/AlertMonitorSummary.tsx index 41193edb3..388d180b2 100644 --- a/apps/dsa-web/src/components/alerts/AlertMonitorSummary.tsx +++ b/apps/dsa-web/src/components/alerts/AlertMonitorSummary.tsx @@ -31,7 +31,7 @@ export const AlertMonitorSummary: React.FC = ({ summary, loading = false ) : null}
@@ -76,7 +76,7 @@ export const AlertMonitorSummary: React.FC = ({ summary, loading = false
- 触发归因只按 rule_id 关联，不按股票 target 猜测规则。 + 触发归因按 rule_id + lifecycle 关联，不按股票 target 猜测规则。
) : null} diff --git a/apps/dsa-web/src/components/alerts/__tests__/AlertMonitorSummary.test.tsx b/apps/dsa-web/src/components/alerts/__tests__/AlertMonitorSummary.test.tsx index 0f95a38a6..5760734e0 100644 --- a/apps/dsa-web/src/components/alerts/__tests__/AlertMonitorSummary.test.tsx +++ b/apps/dsa-web/src/components/alerts/__tests__/AlertMonitorSummary.test.tsx @@ -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(); }); }); diff --git a/docs/alerts.md b/docs/alerts.md index 482ae0bf8..d1371f6db 100644 --- a/docs/alerts.md +++ b/docs/alerts.md @@ -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 边界 diff --git a/src/repositories/alert_repo.py b/src/repositories/alert_repo.py index fd30479a3..2f7476714 100644 --- a/src/repositories/alert_repo.py +++ b/src/repositories/alert_repo.py @@ -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( diff --git a/src/services/alert_worker.py b/src/services/alert_worker.py index 3596585bb..035e251d1 100644 --- a/src/services/alert_worker.py +++ b/src/services/alert_worker.py @@ -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) diff --git a/src/storage.py b/src/storage.py index 18217df9a..b9f5d0395 100644 --- a/src/storage.py +++ b/src/storage.py @@ -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) diff --git a/tests/test_alert_api.py b/tests/test_alert_api.py index e8d4ee9e1..cab65785e 100644 --- a/tests/test_alert_api.py +++ b/tests/test_alert_api.py @@ -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( diff --git a/tests/test_alert_worker.py b/tests/test_alert_worker.py index d47b0348f..c3be2b7ef 100644 --- a/tests/test_alert_worker.py +++ b/tests/test_alert_worker.py @@ -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()