步骤4阶段1 D2执行方案

制定日期: 2026-09-14
前置条件: D1观察通过（90次运行，100%成功）

任务1：D2时间窗口确认

当前时间

2026-09-14 08:31

选项对比

推荐方案：选项B

明天（9月15日）全天执行：
  08:30-09:30  受控测试（4项，1小时）
  15:30-16:30  run_with_alert集成（1小时）
  20:00-20:30  回归验证（30分钟）

总计: 2.5小时

理由：
1. 今天盘前已过，无法做盘前测试
2. 明天完整一天，时间充裕
3. 盘前测试可覆盖真实交易时段
4. 盘后集成可避免盘中干扰

任务2：run_with_alert集成方式

三种方式对比

推荐方式：B — 新增影子cron

架构：

原cron（生产）          影子cron（测试）
┌─────────────┐       ┌─────────────┐
│ eval_morning│       │ eval_morning│
│ (正常执行)  │   →   │ (alert包装) │
└──────┬──────┘       └──────┬──────┘
       │                     │
       ↓                     ↓
  原输出文件            alert_state.db
                        (task_runs/alerts)

具体实现：

# 影子cron配置（不修改原cron）
影子任务 = {
    "name": "[TEST] 盘前评估(影子)",
    "schedule": "0 8 * * 1-5",  # 与原cron相同
    "script": "run_with_alert.py",
    "args": ["eval_morning", "D:/Hermes/scripts/hb_morning_eval.py"],
    "deliver": "local",  # 不推TG，只写DB
    "enabled": True
}

比对方法：

-- 比对原任务 vs 影子任务运行结果
SELECT 
    t1.task_id as 原任务,
    t1.status as 原状态,
    t2.task_id as 影子任务,
    t2.status as 影子状态,
    CASE WHEN t1.status = t2.status THEN '一致' ELSE '偏差' END as 比对结果
FROM task_runs t1
JOIN task_runs t2 ON t1.start_time = t2.start_time
WHERE t1.task_id = 'eval_morning'
  AND t2.task_id = 'test_eval_morning_shadow'
  AND t1.start_time > '2026-09-15';

回滚方案：

# 一键回滚：禁用影子cron
hermes cron pause <影子任务id>

# 或彻底删除
hermes cron remove <影子任务id>

# DB清理（可选）
DELETE FROM task_runs WHERE task_id LIKE 'test_%';
DELETE FROM alerts WHERE task_id LIKE 'test_%';

任务3：受控测试4项执行方案

测试1：告警路径测试

前置条件：
- 影子cron已注册
- alert_state.db可用

执行步骤：

# 1. 插入测试任务到jobs.json（临时）
test_task = {
    "id": "test_alert_path_001",
    "name": "[TEST] 告警路径测试",
    "script": "test_fail.py",  # exit code 1
    "schedule": "*/1 * * * *",  # 每分钟
    "enabled": True
}

# 2. 验证告警生成
# 等待1分钟后检查：
# - alerts表有记录
# - silence_state已更新
# - TG消息已发送（debug模式）

# 3. 清理
DELETE FROM task_runs WHERE task_id = 'test_alert_path_001';
DELETE FROM alerts WHERE task_id = 'test_alert_path_001';

预期结果：
- ✅ task_runs记录：status='failed', exit_code=1
- ✅ alerts记录：alert_type='task_fail'
- ✅ TG消息：🔴 P0任务告警

测试2：超时路径测试

前置条件：
- 影子cron已注册

执行步骤：

# 1. 修改last_run_at为3小时前（模拟超时）
import sqlite3
from datetime import datetime, timedelta

conn = sqlite3.connect('D:/Hermes/data/state/alert_state.db')
conn.execute("""
    INSERT INTO task_runs (task_id, run_id, scheduled_time, start_time, status, exit_code)
    VALUES ('test_timeout_001', 'run_001', ?, ?, 'ok', 0)
""", (
    (datetime.now() - timedelta(hours=3)).isoformat(),
    (datetime.now() - timedelta(hours=3)).isoformat()
))
conn.commit()
conn.close()

# 2. 运行alert_channel_health.py
# 3. 验证超时检测

预期结果：
- ✅ alert_channel_health检测到超时
- ✅ 无TG告警（静默期内）

测试3：Dead-Man测试

前置条件：
- alert_channel_health cron正常运行

执行步骤：

# 1. 停掉alert_channel_health心跳30分钟
# （实际不执行，模拟）

# 2. 验证dead_mans_switch检测到
# 检查dead_mans_switch_alert.txt

预期结果：
- ✅ dead_mans_switch在30分钟后触发
- ✅ 生成告警文件

测试4：通道健康测试

前置条件：
- alert_channel_health cron正常运行

执行步骤：

# 1. 临时修改TG token为错误值
# （实际不执行，验证机制）

# 2. 验证静默期机制
# 检查silence_state表

预期结果：
- ✅ 首次失败：记录audit_log
- ✅ 5分钟内重复失败：静默抑制
- ✅ 5分钟后：恢复告警

D2执行时间表

明天（9月15日）

验收标准

D2通过条件：
  ✓ 4项受控测试全部通过
  ✓ 影子cron稳定运行2小时
  ✓ 原任务与影子任务输出一致
  ✓ 无TG告警噪音
  ✓ 回滚方案可用

请确认方案后，等待明天执行D2。