步骤4阶段1 - 3-4天核心闭环方案（修正版）

制定日期: 2026-09-13
观察任务: cron_id=7e8ae2aeca81
当前进度: D1观察中（17次自动运行，100%成功）

任务1：修正排期表

增量迭代（D5-D7，边跑边做）

任务2：修正受控测试方案（D2）

2.1 测试矩阵

总计耗时: 约2.5小时（可分上午/下午执行）

2.2 测试执行流程

# === 前置：备份生产数据 ===
cp C:/Users/Admin/AppData/Local/hermes/cron/jobs.json C:/Users/Admin/AppData/Local/hermes/cron/jobs.json.bak_D2

# === 测试1：告警路径（30分钟）===
# 1.1 添加测试任务
python -c "
import json
from pathlib import Path
jobs_path = Path('C:/Users/Admin/AppData/Local/hermes/cron/jobs.json')
data = json.loads(jobs_path.read_text(encoding='utf-8'))
data['jobs'].append({
    'id': 'test_alert_path_001',
    'name': '[TEST] 告警路径测试',
    'schedule': '*/15 * * * *',
    'command': 'python -c \"import time; time.sleep(1)\"',
    'last_status': 'failed',
    'last_run_at': '2026-09-14T10:00:00+08:00'
})
jobs_path.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding='utf-8')
print('测试任务已添加')
"

# 1.2 插入failed记录
python -c "
import sqlite3
conn = sqlite3.connect('D:/Hermes/data/state/alert_state.db')
conn.execute(\"\"\"
    INSERT OR REPLACE INTO task_runs 
    (task_id, status, start_time, exit_code, error_message, created_at)
    VALUES ('test_alert_path_001', 'failed', '2026-09-14T10:00:00', 1, 'test alert trigger', datetime('now'))
\"\"\")
conn.commit()
conn.close()
print('测试记录已插入')
"

# 1.3 等待cron运行（约15分钟），检查TG是否收到[TEST]前缀告警

# 1.4 清理测试数据（恢复jobs.json + 删除task_runs记录）

# === 测试2-5：类似流程，分别注入不同故障 ===

2.3 测试通过标准

• [ ] 告警路径：收到包含[TEST]前缀的TG消息
• [ ] 超时路径：收到包含timeout关键词的TG消息
• [ ] Dead-Man：停心跳30分钟后收到dead_man告警
• [ ] 通道健康：改token后收到channel_error告警
• [ ] run_with_alert集成：评估完成后触发告警检查
• [ ] 清理后：jobs.json恢复原状，task_runs无残留

任务3：修正风控熔断边界（D3）

3.1 只做（核心4项）

3.2 不做（非核心）

• ❌ 行业暴露控制
• ❌ 信号失效检测
• ❌ 数据断流监控

3.3 实现方案

# 风控熔断检查脚本（D3实现）
# 文件: D:/Hermes/scripts/risk_circuit_breaker.py

def check_circuit_breaker():
    """核心4项熔断检查"""
    results = {
        "total_position": check_total_position(),      # 总仓位>80%
        "single_stock": check_single_stock_drawdown(), # 单票回撤>5%
        "daily_drawdown": check_daily_drawdown(),      # 单日回撤>3%
        "liquidity": check_liquidity()                 # 流动性不足
    }

    # 输出警告，不阻断主流程
    for item, passed in results.items():
        if not passed:
            log_warning(f"熔断警告: {item}")

    return results

def check_total_position():
    """总仓位检查"""
    # 实现逻辑...
    pass

def check_single_stock_drawdown():
    """单票回撤检查"""
    # 实现逻辑...
    pass

def check_daily_drawdown():
    """单日回撤检查"""
    # 实现逻辑...
    pass

def check_liquidity():
    """流动性检查"""
    # 实现逻辑...
    pass

任务4：修正验收标准

4.1 P0验证（必须完成）

4.2 移除的指标

• ❌ "7天100%成功率" → 改为"3天稳定运行"
• ❌ "cron≤10个" → 移除，不单独考核

任务5：增量迭代计划（D5-D7）

5.1 边跑边做

5.2 不单独排期

• 历史数据回填（低优先级）
• 报告格式优化（低优先级）
• 多源数据融合（长期价值）

附录：当前系统状态

A.1 已部署组件

A.2 D1观察统计

A.3 待执行检查

方案修正时间: 2026-09-13T14:30:00
下一步: 等待20:00晚间检查 + 次日10:00 D1总结后进入D2